一、 前言
今天小編給大家嘮叨下工單審批流的那些事。在運維平臺的建設，少不了工單審批工作流的實現，雖然已經存在大量基于python甚至是django封裝好的庫，而且非常方面，文檔也相對齊全。
但封裝好的庫難免會遇到擴展性差，修改起來麻煩，想增加一些自定義的功能時無從下手。今天給大家簡單介紹下如何利用django實現簡單的審批流，雖然談不上最佳方式，但希望能給大家一些這方面的啟發(fā)。
二、表結構設計
先上一張表結構圖
我們來逐一說明：
workflow：存放所有的工單類型，如發(fā)版申請、電腦故障維修申請、wifi申請等等。
state：存放審批過程中經歷的節(jié)點，如小組長、部門經理、cto等等。
transition：與state表多對多關系，存放當前一個審批節(jié)點與前一個審批節(jié)點之間的關系。其中，同意和拒絕兩種條件，對應兩個不同的下一審批節(jié)點。
state_obj_user：存放每張工單中，審批節(jié)點與審批用戶之間的關系，在創(chuàng)建工單的時候生成。
workflow_state_event：流程事件，每創(chuàng)建一張工單時，以及同意審批使工單進入下一個狀態(tài)時，都會正常一個對應的審批事件記錄，包括審批人、審批時間、審批選項等等，如果審批拒絕，則不生成新的事件記錄。
django_content_type：由于可能存在不止一種申請工單，為了避免每種工單建一套相似的表，可以利用 Django 內置的model映射關系表，其中包含了所有表之間的關系，可以利用該表進行外鍵關聯。
workflow_deveplop_version：發(fā)版申請表，外鍵關聯workflow表，保存該類申請時填寫的一些信息。
三、models代碼部分
from django.db import modelsfrom django.contrib.auth.models import Userfrom django.contrib.contenttypes.fields import GenericForeignKeyfrom django.contrib.contenttypes.models import ContentTypefrom django.contrib.contenttypes.fields import GenericRelation
class Workflow(models.Model): ''' 工單類型表 ''' name = models.CharField(max_length=100, unique=True, verbose_name=u'流程名') abbr = models.CharField(max_length=20, unique=True, default='', verbose_name=u'縮寫') description = models.CharField(max_length=100, default='', verbose_name=u'工單的描述') init_state = models.ForeignKey('State', on_delete=models.SET_NULL, related_name='workflow_init_state', blank=True, null=True, verbose_name=u'初始狀態(tài)')
class Meta: db_table = 'workflow' verbose_name = u'工單類型表' verbose_name_plural = verbose_name
def __str__(self): return self.name
class State(models.Model): '''狀態(tài)表 關聯到對應的流程 常見的狀態(tài): 測試審核，研發(fā)審核，運維審核，完成 ''' name = models.CharField(max_length=100, verbose_name=u'狀態(tài)名') workflow = models.ForeignKey('Workflow', on_delete=models.PROTECT, verbose_name=u'對應的流程名') transition = models.ManyToManyField('Transition', verbose_name=u'狀態(tài)轉化')
class Meta: db_table = 'workflow_state' verbose_name = u'狀態(tài)表' verbose_name_plural = verbose_name
def get_pre_state(self): try: return self.transition.get(condition='拒絕').destination except: return None
def get_latter_state(self): try: return self.transition.get(condition='同意').destination except: return None
def __str__(self): return self.workflow.name + ':' + self.name
class StateObjectUserRelation(models.Model): '''obj和狀態(tài)和的用戶關系 ''' content_type = models.ForeignKey(ContentType, on_delete=models.PROTECT) object_id = models.PositiveIntegerField() content_object = GenericForeignKey('content_type', 'object_id') state = models.ForeignKey('State', on_delete=models.PROTECT, verbose_name=u'關聯狀態(tài)') users = models.ManyToManyField(User, verbose_name=u'關聯用戶')
def __str__(self): return '%s:%s:%s' % (self.content_type.name, self.object_id, self.state.name)
class Meta: unique_together = ('content_type', 'object_id', 'state') db_table = 'state_object_user'
class Transition(models.Model): '''流程轉化,從一個流程轉化到另一個流程 **Attributes:** name 在流程內一個唯一的轉化名稱 workflow 轉化歸屬的流程，必須是一個流程實例 destination 當轉化發(fā)生后的目標指向狀態(tài) condition 發(fā)生轉化的條件 ''' name = models.CharField(max_length=100, verbose_name=u'轉化名稱') workflow = models.ForeignKey('Workflow', on_delete=models.PROTECT, verbose_name=u'所屬的流程') destination = models.ForeignKey('State', on_delete=models.PROTECT, related_name='transition_destination', verbose_name=u'目標狀態(tài)指向') condition = models.CharField(max_length=100, verbose_name=u'發(fā)生轉化的條件')
class Meta: db_table = 'workflow_transition' unique_together = ('workflow', 'name') verbose_name = u'狀態(tài)轉化表' verbose_name_plural = verbose_name
def __str__(self): return self.workflow.name + ':' + self.name
class WorkflowStateEvent(models.Model): '''流程轉化的日志 記錄了每個流程轉化到相應的state時的結果 增加了額外的create_time， creator， title這三個屬性 這三個屬性本來是任意申請的必須字段，他們的值都是相同的 在創(chuàng)建wse的時候，把obj的這三個屬性值賦值過來 這樣做的目的是為了在'我的待審批'和'我的審批記錄'中可以通過關鍵字查找 ''' DING_STATUS = ( (0, '未發(fā)送'), (1, '已發(fā)送'), ) IS_CANCEL = ( (0, '未取消'), (1, '已取消') ) content_type = models.ForeignKey(ContentType, on_delete=models.PROTECT) object_id = models.PositiveIntegerField() content_object = GenericForeignKey('content_type', 'object_id') state = models.ForeignKey('State', blank=True, null=True, on_delete=models.SET_NULL) create_time = models.DateTimeField() approve_time = models.DateTimeField(blank=True, null=True) creator = models.ForeignKey(User, verbose_name=u'工單發(fā)起人', on_delete=models.PROTECT) title = models.CharField(max_length=500, verbose_name=u'標題') is_current = models.BooleanField(default=False, verbose_name=u'是否為當前狀態(tài)') approve_user = models.ForeignKey(User, related_name='approve_user_user', blank=True, null=True, verbose_name=u'審批的用戶', on_delete=models.PROTECT) state_value = models.CharField(max_length=10, blank=True, null=True, verbose_name=u'state的審批值') ding_notice = models.IntegerField(choices=DING_STATUS, default=0, verbose_name=u'是否已經發(fā)送過釘釘通知') opinion = models.CharField(max_length=100, blank=True, null=True, verbose_name=u'審批意見') users = models.ManyToManyField(User, related_name='wse_approve_users', verbose_name=u'指定的審批用戶,每次審批后從sor中copy') is_cancel = models.IntegerField(choices=IS_CANCEL, default=0, verbose_name=u'工單流程是否取消')
class Meta: db_table = 'workflow_state_event' verbose_name = u'流程轉化的日志' verbose_name_plural = verbose_name unique_together = ('content_type', 'object_id', 'state')
def __str__(self): return '%s-%s-%s' % (self.content_object.title, self.state, self.state_value)
def get_current_state_approve_user_list(self): return self.users.all()
def show_apply_history(self): if self.state.name == '完成': state_value = '完成' else: state_value = self.state_value if self.state_value else '審批中'
return { 'wse_id': self.id, 'workflow_id': self.content_object.workflow.id, 'abbr': self.content_object.workflow.abbr, 'workflow': self.content_object.workflow.name, 'title': self.title, 'create_time': str(self.create_time), 'approve_time': str(self.approve_time), 'creator': self.creator.username, 'state': self.state.name, 'state_value': state_value }
def get_opinion(self): return self.opinion if self.opinion else '未填寫'
def get_approve_result(self): if self.state.name == '完成': return '' else: approve_user = self.approve_user.username + ' ' if self.approve_user else ', '.join([u.username for u in self.users.all()]) state_value = self.state_value + ' ' + str(self.approve_time)[:19] if self.state_value else ' 審批中' if state_value == '拒絕': state_value += '，原因：' + self.get_opinion() return approve_user + state_value
class DevelopVersionWorkflow(models.Model): '''發(fā)版申請單''' workflow = models.ForeignKey(Workflow, on_delete=models.PROTECT, verbose_name=u'所屬工作流') create_time = models.DateTimeField(auto_now_add=True, verbose_name=u'創(chuàng)建時間') creator = models.ForeignKey(User, on_delete=models.PROTECT, verbose_name=u'申請人') title = models.CharField(max_length=100, unique=True, verbose_name=u'標題') test_content = models.TextField(default='', verbose_name=u'測試人員填寫內容') dev_content = models.TextField(default='', verbose_name=u'開發(fā)人員填寫內容') code_merge = models.BooleanField(null=True, blank=True, verbose_name=u'代碼是否已合并') wse = GenericRelation(WorkflowStateEvent, related_query_name='develop_version')
class Meta: db_table = 'workflow_develop_version' verbose_name = u'發(fā)版申請單' verbose_name_plural = verbose_name
def __str__(self): return self.workflow.name + '-' + self.title
def is_code_merge(self): if self.code_merge: return '1' elif self.code_merge is None: return '' else: return '0'
四、views代碼部分
主要介紹提交工單與審批工單兩個主要方法類
from django_vue_cmdb.expiring_token_authentication import ExpireTokenAuthenticationfrom rest_framework import permissionsfrom rest_framework.views import APIViewfrom django.http import JsonResponsefrom django.db import IntegrityErrorfrom django.db import transactionfrom django.db.models import Qfrom workflows.models import Workflowfrom workflows.models import DevelopVersionWorkflowfrom workflows.models import WorkflowStateEventfrom workflows.utils import init_workflowfrom workflows.utils import check_approve_permfrom workflows.utils import do_transitionfrom workflows.utils import relate_approve_user_to_wsefrom workflows.utils import get_workflow_chainfrom workflows.utils import get_workflow_chain_with_wsefrom workflows.utils import get_approve_pending_cnt
class WorkflowSubmit(APIView): ''' 工單提交 :param: { 'workflow_id': 1, # 流程id 'apply_form_data': { # 申請表單內容 ... }, } ''' authentication_classes = (ExpireTokenAuthentication,) permission_classes = (permissions.IsAuthenticated,)
def post(self, request, version): result = { 'code': 0, 'msg': '請求成功', 'data': {} } try: with transaction.atomic(): # 校驗操作權限 if not check_user_sso_perm(request.user, 'workflows.workflow_submit'): raise PermissionError pass
raw_data = request.data workflow_abbr = raw_data.get('workflow_abbr', '') workflow = Workflow.objects.get(abbr=workflow_abbr) workflow_abbr = workflow.abbr
# 發(fā)版申請 if workflow_abbr == 'develop_version': title = raw_data.get('title') test_content = raw_data.get('test_content', '') dev_content = raw_data.get('dev_content', '') # 保存申請單內容 obj = DevelopVersionWorkflow.objects.create( title=title, creator=request.user, workflow=workflow, test_content=test_content, dev_content=dev_content ) # 創(chuàng)建流程事件 wse = WorkflowStateEvent.objects.create( content_object=obj, create_time=obj.create_time, creator=request.user, title=obj.title, state=workflow.init_state, is_current=True) # 初始化工單流 init_workflow(workflow, obj, wse)
except PermissionError: result = { 'code': 403, 'msg': '權限受限', 'data': {} } except Workflow.DoesNotExist as e: result = { 'code': 403, 'msg': str(e), 'data': {} } except IntegrityError: result = { 'code': 200, 'msg': '記錄重復', 'data': {} } except Exception as e: result = { 'code': 500, 'msg': str(e), 'data': {} } finally: return JsonResponse(result)
class WorkflowApprove(APIView): ''' 工單審批 :param: { 'wse_id': 2, # 必選，審批的流程事件id 'select': '同意', # 必選，審批選項 'opinion': '意見', # 非必選，審批文字意見 'approve_form_data': { # 非必選，審批表單內容 ... } } ''' authentication_classes = (ExpireTokenAuthentication,) permission_classes = (permissions.IsAuthenticated,)
def post(self, request, version): result = { 'code': 0, 'msg': '請求成功', 'data': {} } try: with transaction.atomic(): approve_user = request.user raw_data = request.data wse_id = raw_data.get('wse_id') wse = WorkflowStateEvent.objects.get(pk=wse_id) dev_content = raw_data.get('dev_content', '') code_merge = raw_data.get('code_merge', '') # 如果有研發(fā)人員填寫內容，則需要保存 if dev_content: wse.content_object.dev_content = dev_content if code_merge: wse.content_object.code_merge = code_merge wse.content_object.save(update_fields=['dev_content', 'code_merge']) # 審批權限檢驗 success, msg = check_approve_perm(wse, approve_user) if not success: raise Exception(msg) # 流程流轉 select = raw_data.get('select') opinion = raw_data.get('opinion', None) success, msg, new_wse = do_transition(wse, select, opinion, approve_user) if success: # 關聯新審批人 relate_approve_user_to_wse(new_wse.state, new_wse.content_object, new_wse) if new_wse.users.all(): # 發(fā)送釘釘通知給下一批審批人員 pass else: # 工單審批完成，繼續(xù)下一步操作 pass else: raise Exception(msg)
except WorkflowStateEvent.DoesNotExist as e: result = { 'code': 500, 'msg': str(e), 'data': {} } except PermissionError: result = { 'code': 403, 'msg': '權限受限', 'data': {} } except IntegrityError: result = { 'code': 200, 'msg': '記錄重復', 'data': {} } except Exception as e: result = { 'code': 500, 'msg': str(e), 'data': {} } finally: return JsonResponse(result)
五、util代碼部分
主要存放一些通用的方法，方面重復使用
# -*- coding: utf-8 -*-from workflows.models import Workflowfrom workflows.models import StateObjectUserRelationfrom workflows.models import WorkflowStateEventfrom django.contrib.contenttypes.models import ContentTypefrom django.contrib.auth.models import Userfrom django.contrib.auth.models import Group
import datetime
def get_approve_user_by_state_name(state_name): ''' 根據審批節(jié)點的名稱獲取審批用戶 如：審批節(jié)點名稱為“測試”，則查找角色名稱為“測試cmdb”下的用戶 ''' sso_role_name = state_name + 'cmdb' sso_role = Group.objects.get(name=sso_role_name) if not sso_role: raise Exception('查詢審批角色 {} 失敗，請聯系管理員！'.format(sso_role_name)) return sso_role.user_set.all()
def recursive_latter_state(curr_state, chain_list): ''' 遞歸獲取后續(xù)審批節(jié)點 return [ { 'state': state_obj1, 'users': [user_obj1, user_obj2] }, { 'state': state_obj2, 'users': [user_obj3, user_obj4] } ] ''' chain_list.append( { 'state': curr_state, 'users': [] if curr_state.name == '完成' else get_approve_user_by_state_name(curr_state.name), 'approve_result': '' } ) if curr_state.get_latter_state(): return recursive_latter_state(curr_state.get_latter_state(), chain_list) else: return chain_list
def get_workflow_chain(workflow_id): '''生成審批鏈''' workflow = Workflow.objects.get(pk=workflow_id) chain_list = [] return recursive_latter_state(curr_state=workflow.init_state, chain_list=chain_list)
def get_sor(state, obj): '''根據state和obj從StateObjectUserRelation中獲取一條記錄'''
ctype = ContentType.objects.get_for_model(obj) try: sor = StateObjectUserRelation.objects.get(content_type=ctype, object_id=obj.id, state=state) except StateObjectUserRelation.DoesNotExist: return None return sor
def init_workflow(workflow, obj, wse): ''' 初始化工單： 1. 創(chuàng)建工單整個生命周期經歷的狀態(tài)鏈，及每個狀態(tài)對應的審批人 2. 關聯初始審批節(jié)點的審批人 ''' # 創(chuàng)建工單整個生命周期經歷的狀態(tài)鏈，及每個狀態(tài)對應的審批人 chain_list = get_workflow_chain(workflow.id) for chain in chain_list: sor = StateObjectUserRelation.objects.create(content_object=obj, state=chain['state']) if chain['state'].name != '完成': sor.users.add(*chain['users']) # 關聯初始審批節(jié)點的審批人 relate_approve_user_to_wse(state=workflow.init_state, obj=obj, wse=wse)
def relate_approve_user_to_wse(state, obj, wse): '''審批事件關聯審批用戶''' sor = get_sor(state=state, obj=obj) if sor: users = tuple(sor.users.all()) wse.users.add(*users)
def check_approve_perm(wse, approve_user): '''檢查流程事件當前狀態(tài)是否允許審批，提交審批人是否有權限審批''' if not wse.is_current: return False, '流程事件wse id = {}，當前狀態(tài)不允許審批'.format(wse.id) if approve_user not in wse.users.all(): return False, '您沒有權限審批' return True, '檢查通過'
def get_approved_user(obj, next_state_user): '''根據workflow和申請的obj 從state或者sor中獲取所有的需要審批的用戶，如果之前的用戶已經審批過，返回用戶 不然，返回None '''
ctype = ContentType.objects.get_for_model(obj) list_wse = WorkflowStateEvent.objects.filter(content_type=ctype, object_id=obj.id) for wse in list_wse: if wse.approve_user in next_state_user: return wse.approve_user return None
def do_transition(wse, select, opinion, approve_user): '''流程流轉''' success = True msg = 'ok' new_wse = '' try: if select == '同意': # 創(chuàng)建新的流程事件 transition = wse.state.transition.get(condition=select) new_wse = WorkflowStateEvent.objects.create(content_object=wse.content_object, state=transition.destination, create_time=wse.create_time, creator=wse.creator, title=wse.title, is_current=True, opinion=opinion) # 當前的流程事件設置為已經審批過 wse.is_current = False wse.state_value = transition.condition wse.approve_user = approve_user wse.approve_time = datetime.datetime.now() wse.opinion = opinion wse.save()
# 如果下一個審批節(jié)點的審批用戶還是自己 # 或者是下一個節(jié)點審批人是工單發(fā)起人自身 # 或者是下一個節(jié)點的審批用戶已經審批過之前的節(jié)點 next_state_user = [] sor = get_sor(state=new_wse.state, obj=new_wse.content_object) if sor: next_state_user = sor.users.all()
if approve_user in next_state_user: success, msg, new_wse = do_transition(new_wse, select, opinion, approve_user) elif new_wse.creator in next_state_user: success, msg, new_wse = do_transition(new_wse, select, opinion, new_wse.creator) else: approved_user = get_approved_user(new_wse.content_object, next_state_user) if approved_user: success, msg, new_wse = do_transition(new_wse, select, opinion, approve_user)
elif select == '拒絕': # 如果拒絕，則流程終止，不再創(chuàng)建新的流程事件 wse.approve_user = approve_user wse.state_value = select wse.opinion = opinion wse.approve_time = datetime.datetime.now() wse.save() new_wse = wse else: raise Exception('未知的審批選項')
except Exception as e: success = False msg = str(e) finally: return success, msg, new_wse
def get_workflow_chain_with_wse(wse): '''根據流程事件wse生成審批鏈，包括每個審批節(jié)點的審批選項（同意、拒絕、審批中）''' wse_objs = wse.content_object.wse.all() active = wse_objs.count() # 當前的審批進度鏈 current_chain = [{'state': wse.state, 'users': wse.users.all(), 'approve_result': wse.get_approve_result()} for wse in wse_objs] # 整個審批過程的審批鏈 common_chain = get_workflow_chain(wse.content_object.workflow.id) if len(current_chain) == len(common_chain): return active, current_chain # 當前審批鏈與整個審批鏈進行對比 for curr_ch in current_chain: for com_ch in common_chain: if curr_ch['state'] == com_ch['state']: com_ch['approve_result'] = curr_ch['approve_result'] break return active, common_chain
def get_approve_pending_cnt(user): '''獲取用戶的待審批工單數量''' pending_cnt = len([wse for wse in WorkflowStateEvent.objects.filter(is_current=True) if user in wse.get_current_state_approve_user_list() and wse.state_value is None]) return pending_cnt
關于工單流轉的過程：
關于通用外鍵的使用，可以參考：
https://docs.djangoproject.com/en/3.0/ref/contrib/contenttypes/
項目參考：
https://github.com/CJFJack/django_vue_cmdb
