简介：这是一套基于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_length=100, unique=True) code = models.SlugField(max_length=50, unique=True) # 供代码引用的稳定标识 version = models.PositiveIntegerField(default=1) is_active = models.BooleanField(default=True) created_at = models.DateTimeField(auto_now_add=True) class Meta: unique_together = ("code", "version") class NodeDef(models.Model): """节点定义：流程里的一个环节，比如'主管审批'""" NODE_TYPE = (("start", "开始"), ("task", "审批任务"), ("end", "结束")) workflow = models.ForeignKey(WorkflowDef, on_delete=models.CASCADE, related_name="nodes") name = models.CharField(max_length=100) node_type = models.CharField(max_length=10, choices=NODE_TYPE) # 审批人解析规则：可以是角色、指定用户、或工单上的某个字段 assignee_type = models.CharField(max_length=20, default="role") assignee_value = models.CharField(max_length=100, blank=True) order = models.PositiveIntegerField(default=0) # 仅用于展示排序 class Transition(models.Model): """流转规则：从 from_node 到 to_node 的条件""" workflow = models.ForeignKey(WorkflowDef, on_delete=models.CASCADE, related_name="transitions") from_node = models.ForeignKey(NodeDef, on_delete=models.CASCADE, related_name="outgoing") to_node = models.ForeignKey(NodeDef, on_delete=models.CASCADE, related_name="incoming") # 条件表达式，比如 amount > 5000，用受限的表达式求值 condition = models.CharField(max_length=255, blank=True) priority = models.IntegerField(default=0) # 多条规则命中时取优先级高的 class WorkflowInstance(models.Model): """流程实例：某张工单走某条流程的运行态""" workflow = models.ForeignKey(WorkflowDef, on_delete=models.PROTECT) current_node = models.ForeignKey(NodeDef, on_delete=models.PROTECT, null=True) biz_type = models.CharField(max_length=50) # 关联的业务类型，如 'ticket' biz_id = models.PositiveIntegerField() # 关联的业务主键 status = models.CharField(max_length=20, default="running") # running/finished/cancelled created_at = models.DateTimeField(auto_now_add=True) class Meta: indexes = [models.Index(fields=["biz_type", "biz_id"])] class Task(models.Model): """任务：实例在某个节点上产生的待办""" instance = models.ForeignKey(WorkflowInstance, on_delete=models.CASCADE, related_name="tasks") node = models.ForeignKey(NodeDef, on_delete=models.PROTECT) assignee = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.PROTECT, null=True) status = models.CharField(max_length=20, default="pending") # pending/approved/rejected comment = models.TextField(blank=True) created_at = models.DateTimeField(auto_now_add=True) finished_at = models.DateTimeField(null=True, blank=True)

逻辑说明：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(id=instance_id) if inst.status != "running": raise ValueError("流程已结束，不能推进") current = inst.current_node # 找到所有从当前节点出发的规则，按优先级排序 rules = Transition.objects.filter( workflow=inst.workflow, from_node=current ).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(instance=inst, node=current, status="pending").update( status="approved" if action == "approve" else "rejected", comment=comment, 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(instance=inst, node=next_node, assignee=assignee) 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(code=workflow_code, is_active=True).latest("version") start_node = wf.nodes.get(node_type="start") inst = WorkflowInstance.objects.create( workflow=wf, current_node=start_node, biz_type="ticket", biz_id=ticket.id, ) # 开始节点自动流转到第一个任务节点 first_task_node = wf.nodes.filter(node_type="task").order_by("order").first() if first_task_node: inst.current_node = first_task_node inst.save() Task.objects.create( instance=inst, node=first_task_node, assignee=resolve_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( assignee=request.user, status="pending" ).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(id=task_id, assignee=request.user, status="pending") action = request.POST.get("action", "approve") advance(task.instance_id, action, request.user, request.POST.get("comment", "")) return JsonResponse({"ok": True})

逻辑说明：select_related一次性把关联的实例和节点查出来，避免待办列表 N+1 查询，这是列表页性能的关键。审批接口先校验任务归属和状态，防止越权审批别人的单子。

参数说明：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触发 N+1 查询。解决：查询时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, names=context)) except NameNotDefined as e: # 条件引用了工单上不存在的字段，直接报错比静默通过安全 raise ValueError(f"条件引用了未知字段: {e}")

逻辑说明：simple_eval默认禁用函数调用和属性访问，只允许基本运算和名字查找，比eval安全。names=context把工单字段注入进去，表达式里直接写amount > 5000 and dept == 'finance'。参数说明：context里的值类型要和表达式里的字面量匹配，字符串比较记得加引号，数字别加。

第二个是流程可视化。运营最怕配完流程不知道长啥样。我一般加一个只读接口，把节点和流转规则吐成 JSON，前端用图形库画出来。核心是别让运营去猜priority谁大谁小，图上用箭头粗细或颜色标出优先级。

# views.py 片段 def workflow_graph(request, code): wf = WorkflowDef.objects.filter(code=code, is_active=True).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、类型不一致，线上翻车一次排查半天。宁可多打几行日志，也别省这一步。希望帮到你。

本文还有配套的精品资源，点击获取