简介这是一套基于Django框架实现的工作流引擎与工单系统完整项目源码面向Python Web开发学习者、软件工程专业学生及需要业务流程管理方案的开发者尤其适合作为毕业设计选题。项目以Python为主要语言围绕工作流定义、任务分配、多级审批、状态跟踪与自定义流程等核心能力展开帮助读者理解工单从创建到闭环的完整业务链路。压缩包共362个文件约17.06MB其中80个py文件承载后端模型与视图逻辑58个tsx与47个ts文件构成前端界面另含55张png、12张jpg等界面截图与演示素材以及sql、yml、Dockerfile、md等配置与说明文档目录结构清晰便于按模块检索。目前已有360人学习下载。通过研读源码与配套说明读者可掌握Django的MTV设计模式、工作流引擎的实现思路与部署方法积累需求分析、设计、实现到测试的完整工程经验对提升Web开发实战能力有切实帮助。1. 从一张工单卡三天说起Django 工作流引擎到底解决什么问题我见过太多团队用 Django 做后台工单表建得漂漂亮亮状态字段就一个status值从pending到done硬编码在视图里。结果业务一改流程——请假要先组长审再 HR 审采购要加一道财务复核——开发就得改代码、改迁移、重新上线。一张工单在三个部门之间卡三天谁都不知道卡在谁那里因为状态流转全散落在if/else里没有一张图能说清楚。基于 Django 的工作流引擎加工单系统要解决的就是这件事把「谁在什么条件下把工单推到下一步」从代码里抽出来变成可配置的数据。工单是业务对象工作流引擎是驱动它流转的规则机。这套东西适合谁适合正在用 Django 做 OA、审批、售后派工、生产报工这类系统的后端同学尤其是那种流程会变、节点会加、审批人会换但你又不想每次改流程都发版的场景。Python 生态里做工作流的选择不少但如果你已经在 Django 上用它的 ORM、Admin、权限体系去承载工作流落地成本最低。2. 工作流引擎的数据模型怎么设计五张表撑起流转2.1 为什么不用状态字段硬扛而要拆成定义与实例先说选型理由。工作流引擎的核心矛盾是「流程定义」和「流程实例」必须分开。定义是模板比如「报销流程提交 → 主管审批 → 财务审批 → 归档」实例是某一张具体工单走这条流程时产生的运行记录。如果你把两者混在一张表里改一次模板就会污染所有历史工单的流转记录这是血泪经验。常见做法是拆成五张表流程定义、节点定义、流转规则、流程实例、任务。下面是我一般会用的模型骨架字段做了精简实际项目按需加。# models.py from django.db import models from django.conf import settings class WorkflowDef(models.Model): 流程定义一个模板比如报销流程 name models.CharField(max_length100, uniqueTrue) code models.SlugField(max_length50, uniqueTrue) # 供代码引用的稳定标识 version models.PositiveIntegerField(default1) is_active models.BooleanField(defaultTrue) created_at models.DateTimeField(auto_now_addTrue) class Meta: unique_together (code, version) class NodeDef(models.Model): 节点定义流程里的一个环节比如主管审批 NODE_TYPE ((start, 开始), (task, 审批任务), (end, 结束)) workflow models.ForeignKey(WorkflowDef, on_deletemodels.CASCADE, related_namenodes) name models.CharField(max_length100) node_type models.CharField(max_length10, choicesNODE_TYPE) # 审批人解析规则可以是角色、指定用户、或工单上的某个字段 assignee_type models.CharField(max_length20, defaultrole) assignee_value models.CharField(max_length100, blankTrue) order models.PositiveIntegerField(default0) # 仅用于展示排序 class Transition(models.Model): 流转规则从 from_node 到 to_node 的条件 workflow models.ForeignKey(WorkflowDef, on_deletemodels.CASCADE, related_nametransitions) from_node models.ForeignKey(NodeDef, on_deletemodels.CASCADE, related_nameoutgoing) to_node models.ForeignKey(NodeDef, on_deletemodels.CASCADE, related_nameincoming) # 条件表达式比如 amount 5000用受限的表达式求值 condition models.CharField(max_length255, blankTrue) priority models.IntegerField(default0) # 多条规则命中时取优先级高的 class WorkflowInstance(models.Model): 流程实例某张工单走某条流程的运行态 workflow models.ForeignKey(WorkflowDef, on_deletemodels.PROTECT) current_node models.ForeignKey(NodeDef, on_deletemodels.PROTECT, nullTrue) biz_type models.CharField(max_length50) # 关联的业务类型如 ticket biz_id models.PositiveIntegerField() # 关联的业务主键 status models.CharField(max_length20, defaultrunning) # running/finished/cancelled created_at models.DateTimeField(auto_now_addTrue) class Meta: indexes [models.Index(fields[biz_type, biz_id])] class Task(models.Model): 任务实例在某个节点上产生的待办 instance models.ForeignKey(WorkflowInstance, on_deletemodels.CASCADE, related_nametasks) node models.ForeignKey(NodeDef, on_deletemodels.PROTECT) assignee models.ForeignKey(settings.AUTH_USER_MODEL, on_deletemodels.PROTECT, nullTrue) status models.CharField(max_length20, defaultpending) # pending/approved/rejected comment models.TextField(blankTrue) created_at models.DateTimeField(auto_now_addTrue) finished_at models.DateTimeField(nullTrue, blankTrue)逻辑说明WorkflowDef用code version做唯一约束是为了让流程升级时旧实例还能指向旧版本不至于历史记录错乱。Transition里的condition存字符串而不是 Python 代码是为了安全——绝不要用eval直接跑用户输入。WorkflowInstance用biz_type biz_id做多态关联这样一套引擎能同时驱动工单、请假、采购多种业务不用给每种业务建一套流程表。参数说明assignee_type我一般支持三种值——role按角色找、user指定人、field从工单字段取比如工单上的owner。priority越大越优先避免多条流转规则同时命中时结果不确定。2.2 用 Django 迁移把表建起来并挂进 Admin模型写完先跑迁移。这一步新手最容易翻车的地方是忘了INSTALLED_APPS里注册 app报No installed app with label。# 假设 app 名叫 workflow python manage.py makemigrations workflow python manage.py migrate逻辑说明makemigrations根据模型差异生成迁移文件migrate把变更应用到数据库。如果你改了模型字段又不想丢数据别急着删迁移文件重来先看migrate的报错多数是字段非空但没默认值。参数说明生产环境我一般会先python manage.py sqlmigrate workflow 0001看一眼生成的 SQL确认没有意外的全表锁操作再执行migrate。把定义类模型挂进 Admin运营就能自己配流程不用开发介入# admin.py from django.contrib import admin from .models import WorkflowDef, NodeDef, Transition, WorkflowInstance, Task class NodeInline(admin.TabularInline): model NodeDef extra 1 class TransitionInline(admin.TabularInline): model Transition extra 1 fk_name workflow admin.register(WorkflowDef) class WorkflowDefAdmin(admin.ModelAdmin): list_display (name, code, version, is_active) inlines [NodeInline, TransitionInline] admin.register(Task) class TaskAdmin(admin.ModelAdmin): list_display (instance, node, assignee, status, created_at) list_filter (status, node)逻辑说明TabularInline让节点和流转规则在流程定义页面里直接编辑运营配流程像填表格。TaskAdmin的list_filter让审批人按状态筛自己的待办。参数说明extra 1控制默认空行数别设太大否则页面加载慢。fk_name在多个外键指向同一模型时必须指定否则 Django 不知道用哪个。3. 流转逻辑怎么写从提交工单到自动推进节点3.1 引擎核心一个函数负责「当前节点 动作 → 下一节点」引擎的心脏是一个纯函数给定实例当前节点和一个动作同意/驳回根据流转规则算出下一个节点并生成对应任务。把它写成独立模块别塞进视图方便单测。# engine.py from django.db import transaction from .models import WorkflowInstance, NodeDef, Transition, Task def _eval_condition(expr, context): 安全求值只允许比较运算禁止函数调用和属性访问 if not expr: return True # 只放行数字、比较符、and/or/not 和上下文字段名 allowed set(0123456789. !()andornot_) if not set(expr) allowed: raise ValueError(f非法条件表达式: {expr}) return bool(eval(expr, {__builtins__: {}}, context)) transaction.atomic def advance(instance_id, action, user, comment): 推进流程action 为 approve 或 reject inst WorkflowInstance.objects.select_for_update().get(idinstance_id) if inst.status ! running: raise ValueError(流程已结束不能推进) current inst.current_node # 找到所有从当前节点出发的规则按优先级排序 rules Transition.objects.filter( workflowinst.workflow, from_nodecurrent ).order_by(-priority) # 构造条件上下文把工单字段和动作都放进去 context {action: action, amount: getattr(inst, amount, 0)} next_node None for rule in rules: if _eval_condition(rule.condition, context): next_node rule.to_node break if next_node is None: raise ValueError(f节点 {current.name} 没有匹配的流转规则) # 关闭当前任务 Task.objects.filter(instanceinst, nodecurrent, statuspending).update( statusapproved if action approve else rejected, commentcomment, finished_at__import__(django.utils.timezone, fromlist[now]).now(), ) # 更新实例并生成新任务 inst.current_node next_node if next_node.node_type end: inst.status finished inst.save() if next_node.node_type task: assignee resolve_assignee(next_node, inst) Task.objects.create(instanceinst, nodenext_node, assigneeassignee) return inst逻辑说明select_for_update加行锁防止两个审批人同时点同意导致流程走两遍这是并发场景下的后悔药。_eval_condition用白名单字符集过滤表达式比直接eval安全得多但仍然建议生产环境换成asteval这类库。流转规则按priority降序第一条命中的生效所以「金额大于 5000 走财务」这种规则要设更高优先级。参数说明context里放什么取决于你的条件表达式要引用哪些字段。我一般把工单的关键字段金额、类型、部门都塞进去。resolve_assignee是审批人解析函数按assignee_type分支处理这里省略实现。3.2 提交工单时启动流程实例工单创建后要自动挂上流程。常见做法是在工单模型的save后或视图里调用启动函数。# services.py from django.db import transaction from .models import WorkflowDef, WorkflowInstance, NodeDef, Task transaction.atomic def start_workflow(ticket, workflow_code): 为工单启动一条流程返回实例 wf WorkflowDef.objects.filter(codeworkflow_code, is_activeTrue).latest(version) start_node wf.nodes.get(node_typestart) inst WorkflowInstance.objects.create( workflowwf, current_nodestart_node, biz_typeticket, biz_idticket.id, ) # 开始节点自动流转到第一个任务节点 first_task_node wf.nodes.filter(node_typetask).order_by(order).first() if first_task_node: inst.current_node first_task_node inst.save() Task.objects.create( instanceinst, nodefirst_task_node, assigneeresolve_assignee(first_task_node, inst), ) return inst逻辑说明latest(version)取最新版本的流程定义保证新工单走新流程老工单不受影响。开始节点不产生待办直接跳到第一个任务节点这样审批人一进来就能看到活。参数说明workflow_code建议从工单类型映射过来比如售后工单对应after_sale别硬编码在视图里用一张配置表或常量映射。3.3 待办列表和审批接口审批人需要看到自己的待办点同意/驳回触发advance。# views.py from django.contrib.auth.decorators import login_required from django.http import JsonResponse from django.views.decorators.http import require_POST from .models import Task from .engine import advance login_required def my_tasks(request): tasks Task.objects.filter( assigneerequest.user, statuspending ).select_related(instance, node) data [ {task_id: t.id, node: t.node.name, biz_id: t.instance.biz_id} for t in tasks ] return JsonResponse({tasks: data}) login_required require_POST def approve_task(request, task_id): task Task.objects.get(idtask_id, assigneerequest.user, statuspending) action request.POST.get(action, approve) advance(task.instance_id, action, request.user, request.POST.get(comment, )) return JsonResponse({ok: True})逻辑说明select_related一次性把关联的实例和节点查出来避免待办列表 N1 查询这是列表页性能的关键。审批接口先校验任务归属和状态防止越权审批别人的单子。参数说明action从请求体取默认approve。真实项目里驳回往往要带原因comment字段必填校验要加上。4. 避坑与排查工单流转里最容易翻车的五件事4.1 现象两个审批人同时点同意流程走了两遍原因advance里没加行锁两个请求同时读到current_node还是旧节点各自算了一遍下一节点生成两条任务。解决用select_for_update()锁住实例行并且把整个推进逻辑包在transaction.atomic里。注意select_for_update必须在事务内才生效单独调用没用。4.2 现象改了流程定义历史工单的流转记录全乱了原因流程定义和实例共用一份节点数据改模板等于改历史。解决定义表加version实例创建时锁定当时的workflow外键新版本另起一条WorkflowDef记录。永远不要原地修改已被实例引用的流程定义。4.3 现象条件表达式报NameError或执行了危险代码原因用eval直接跑用户配置的字符串用户写了__import__(os).system(...)。解决白名单字符集过滤或者换asteval、simpleeval这类受限求值库。我一般还会在保存Transition时做一次语法校验把非法表达式挡在配置阶段。4.4 现象待办列表越加载越慢几百条就要好几秒原因循环里访问task.instance.biz_id触发 N1 查询。解决查询时select_related(instance, node)如果还要展示工单详情用prefetch_related或批量查一次工单再在内存里拼。用 Django Debug Toolbar 看一眼 SQL 条数超过 10 条基本就是这个问题。4.5 现象审批人离职了任务卡在他名下没人能处理原因assignee是硬关联到用户人一走任务就成孤儿。解决加一个「转办」接口允许管理员把pending任务改派或者assignee_type用role任务不绑具体人绑角色谁有这个角色谁就能审。后者更抗人员变动但要注意角色里多个人时的抢单问题需要再加一层认领逻辑。5. 让引擎真正好用条件表达式进阶与流程可视化前面搭的是能跑的最小闭环但真上线你会发现两个需求条件越来越复杂运营看不懂流程长什么样。这一章说两个具体技巧。第一个是条件表达式的进阶。纯字符串比较不够用比如「金额大于 5000 且部门是财务」。我一般把condition设计成支持and/or/not和字段引用的表达式配合受限求值库。下面是一个用simpleeval的写法比手写白名单稳# engine.py 片段 from simpleeval import simple_eval, NameNotDefined def _eval_condition(expr, context): if not expr: return True try: return bool(simple_eval(expr, namescontext)) except NameNotDefined as e: # 条件引用了工单上不存在的字段直接报错比静默通过安全 raise ValueError(f条件引用了未知字段: {e})逻辑说明simple_eval默认禁用函数调用和属性访问只允许基本运算和名字查找比eval安全。namescontext把工单字段注入进去表达式里直接写amount 5000 and dept finance。参数说明context里的值类型要和表达式里的字面量匹配字符串比较记得加引号数字别加。第二个是流程可视化。运营最怕配完流程不知道长啥样。我一般加一个只读接口把节点和流转规则吐成 JSON前端用图形库画出来。核心是别让运营去猜priority谁大谁小图上用箭头粗细或颜色标出优先级。# views.py 片段 def workflow_graph(request, code): wf WorkflowDef.objects.filter(codecode, is_activeTrue).latest(version) nodes [{id: n.id, name: n.name, type: n.node_type} for n in wf.nodes.all()] edges [ {from: t.from_node_id, to: t.to_node_id, condition: t.condition, priority: t.priority} for t in wf.transitions.all() ] return JsonResponse({nodes: nodes, edges: edges})逻辑说明这个接口只读不涉及权限敏感数据但仍要login_required。前端拿到nodes和edges就能渲染有向图condition显示在箭头上运营一眼能看出哪条规则先命中。验证方法上我习惯给引擎写一组单测覆盖「正常流转」「条件不命中」「并发推进」「驳回」四种情况。用 Django 的TestCase加transaction回滚跑一遍几秒钟比手工点页面靠谱得多。具体做法是造一条流程定义建一个实例调advance断言current_node和Task数量符合预期。最后说个我自己的习惯每次改流转逻辑先在本地用manage.py shell手动跑一遍advance把context打出来看条件求值结果确认无误再写进代码。工作流这东西玄学都藏在条件表达式的边界情况里比如空字符串、None、类型不一致线上翻车一次排查半天。宁可多打几行日志也别省这一步。希望帮到你。本文还有配套的精品资源点击获取