基于 Django 的工单审批流实现
一、 前言
今天小编给大家唠叨下工单审批流的那些事。在运维平台的建设，少不了工单审批工作流的实现，虽然已经存在大量基于python甚至是django封装好的库，而且非常方面，文档也相对齐全。
但封装好的库难免会遇到扩展性差，修改起来麻烦，想增加一些自定义的功能时无从下手。今天给大家简单介绍下如何利用django实现简单的审批流，虽然谈不上最佳方式，但希望能给大家一些这方面的启发。
二、表结构设计
先上一张表结构图
我们来逐一说明：
workflow：存放所有的工单类型，如发版申请、电脑故障维修申请、wifi申请等等。
state：存放审批过程中经历的节点，如小组长、部门经理、cto等等。
transition：与state表多对多关系，存放当前一个审批节点与前一个审批节点之间的关系。其中，同意和拒绝两种条件，对应两个不同的下一审批节点。
state_obj_user：存放每张工单中，审批节点与审批用户之间的关系，在创建工单的时候生成。
workflow_state_event：流程事件，每创建一张工单时，以及同意审批使工单进入下一个状态时，都会正常一个对应的审批事件记录，包括审批人、审批时间、审批选项等等，如果审批拒绝，则不生成新的事件记录。
django_content_type：由于可能存在不止一种申请工单，为了避免每种工单建一套相似的表，可以利用 Django 内置的model映射关系表，其中包含了所有表之间的关系，可以利用该表进行外键关联。
workflow_deveplop_version：发版申请表，外键关联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'初始状态')
class Meta: db_table = 'workflow' verbose_name = u'工单类型表' verbose_name_plural = verbose_name
def __str__(self): return self.name
class State(models.Model): '''状态表 关联到对应的流程 常见的状态: 测试审核，研发审核，运维审核，完成 ''' name = models.CharField(max_length=100, verbose_name=u'状态名') workflow = models.ForeignKey('Workflow', on_delete=models.PROTECT, verbose_name=u'对应的流程名') transition = models.ManyToManyField('Transition', verbose_name=u'状态转化')
class Meta: db_table = 'workflow_state' verbose_name = u'状态表' 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和状态和的用户关系 ''' 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'关联状态') 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 当转化发生后的目标指向状态 condition 发生转化的条件 ''' 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'目标状态指向') condition = models.CharField(max_length=100, verbose_name=u'发生转化的条件')
class Meta: db_table = 'workflow_transition' unique_together = ('workflow', 'name') verbose_name = u'状态转化表' verbose_name_plural = verbose_name
def __str__(self): return self.workflow.name + ':' + self.name
class WorkflowStateEvent(models.Model): '''流程转化的日志 记录了每个流程转化到相应的state时的结果 增加了额外的create_time， creator， title这三个属性 这三个属性本来是任意申请的必须字段，他们的值都是相同的 在创建wse的时候，把obj的这三个属性值赋值过来 这样做的目的是为了在'我的待审批'和'我的审批记录'中可以通过关键字查找 ''' DING_STATUS = ( (0, '未发送'), (1, '已发送'), ) 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'工单发起人', on_delete=models.PROTECT) title = models.CharField(max_length=500, verbose_name=u'标题') is_current = models.BooleanField(default=False, verbose_name=u'是否为当前状态') 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'是否已经发送过钉钉通知') 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): '''发版申请单''' workflow = models.ForeignKey(Workflow, on_delete=models.PROTECT, verbose_name=u'所属工作流') create_time = models.DateTimeField(auto_now_add=True, verbose_name=u'创建时间') 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'开发人员填写内容') 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'发版申请单' 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
# 发版申请 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 ) # 创建流程事件 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', '') # 如果有研发人员填写内容，则需要保存 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(): # 发送钉钉通知给下一批审批人员 pass else: # 工单审批完成，继续下一步操作 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): ''' 根据审批节点的名称获取审批用户 如：审批节点名称为“测试”，则查找角色名称为“测试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): ''' 递归获取后续审批节点 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. 创建工单整个生命周期经历的状态链，及每个状态对应的审批人 2. 关联初始审批节点的审批人 ''' # 创建工单整个生命周期经历的状态链，及每个状态对应的审批人 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']) # 关联初始审批节点的审批人 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): '''检查流程事件当前状态是否允许审批，提交审批人是否有权限审批''' if not wse.is_current: return False, '流程事件wse id = {}，当前状态不允许审批'.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 == '同意': # 创建新的流程事件 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()
# 如果下一个审批节点的审批用户还是自己 # 或者是下一个节点审批人是工单发起人自身 # 或者是下一个节点的审批用户已经审批过之前的节点 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 == '拒绝': # 如果拒绝，则流程终止，不再创建新的流程事件 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生成审批链，包括每个审批节点的审批选项（同意、拒绝、审批中）''' 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