82 KiB
工作流模块(2.1.11 流程管理)实施计划
For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (
- [ ]) syntax for tracking.
Goal: 实现完整的自研工作流系统——包含流程配置管理、AntV X6图形化设计器、待办中心、已办中心、后端图遍历引擎。
Architecture: 后端图遍历引擎读取 wf_node + wf_transition 关系表执行节点分发。前端 AntV X6 设计器保存 graph_json(设计器快照)+ 同步写入 wf_node/wf_transition(运行时数据)。业务模块通过 API 调用发起流程,引擎通过回调通知业务状态变更。
Tech Stack: FastAPI, SQLAlchemy, Pydantic v2, Vue 3, Element Plus, AntV X6, TypeScript
Design Spec: docs/superpowers/specs/2026-05-13-workflow-module-design.md
File Structure
Backend(新建文件)
backend/app/models/models.py— 新增6个wf_*模型(追加到现有文件)backend/app/schemas/workflow.py— Pydantic请求/响应schemasbackend/app/crud/workflow.py— CRUD操作backend/app/services/workflow_engine.py— 核心引擎(节点分发、条件评估、审批人解析)backend/app/services/workflow_callback.py— 回调服务backend/app/api/workflow.py— API路由backend/app/api/__init__.py— 注册workflow路由(修改)backend/app/main.py— 注册workflow路由(修改)docs/sql/zkxw.sql— 数据库全量导出基线;增量 schema 可先落库再重导或另附docs/sql/补丁(如callback_url字段)
Frontend(新建文件)
web-admin/src/api/workflow.ts— API调用web-admin/src/types/api.d.ts— 新增Workflow命名空间(追加)web-admin/src/views/system/workflow/process/index.vue— 流程配置管理web-admin/src/views/system/workflow/designer/FlowDesigner.vue— AntV X6设计器web-admin/src/views/system/workflow/task-pending/index.vue— 待办中心web-admin/src/views/system/workflow/task-completed/index.vue— 已办中心web-admin/src/views/system/workflow/components/ApprovalDialog.vue— 审批弹窗web-admin/src/views/system/workflow/components/ProcessStatusTag.vue— 状态标签web-admin/src/router/modules/sqxp.ts— 新增3个路由(修改)web-admin/src/App.vue— 标题白名单(修改)
SQL数据
docs/sql/zkxw.sql— 全库导出(含sys_menu等);若仅需菜单补丁可在docs/sql/另增增量脚本
Phase 1: Backend Foundation
Task 1: SQLAlchemy Models(6个wf_*模型)
Files:
-
Modify:
backend/app/models/models.py(追加到文件末尾) -
*Step 1: 在 models.py 末尾追加6个wf_模型
在文件末尾追加以下代码。字段顺序遵循规范:id → 业务字段 → tenant_no → order_num → remark → create_by → create_dept_id → create_time → update_by → update_time → del_flag。
# =============================================================================
# 工作流(2.1.11 流程管理)
# =============================================================================
class WfProcessDef(Base):
"""流程定义 wf_process_def"""
__tablename__ = "wf_process_def"
id = Column(String(64), primary_key=True, comment="主键ID")
process_code = Column(String(64), nullable=False, comment="流程编码")
process_name = Column(String(200), nullable=False, comment="流程名称")
status = Column(CHAR(1), nullable=False, default="1", comment="1启用 0停用")
tenant_no = Column(String(64), nullable=False, default="zkxw", comment="租户编号")
order_num = Column(Integer, nullable=True, comment="排序号")
remark = Column(String(512), nullable=True, comment="备注")
create_by = Column(String(64), nullable=True, comment="创建人id")
create_dept_id = Column(String(64), nullable=False, default="", comment="创建人部门ID")
create_time = Column(DateTime, nullable=False, default=datetime.now, comment="创建时间")
update_by = Column(String(64), nullable=True, comment="更新人id")
update_time = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now, comment="更新时间")
del_flag = Column(SmallInteger, nullable=False, default=0, comment="删除标记 0正常 1已删除")
class WfProcessVersion(Base):
"""流程定义版本 wf_process_version"""
__tablename__ = "wf_process_version"
id = Column(String(64), primary_key=True, comment="主键ID")
process_def_id = Column(String(64), nullable=False, comment="流程定义ID")
version_no = Column(Integer, nullable=False, comment="版本号,递增")
graph_json = Column(Text, nullable=True, comment="节点与连线等配置(JSON字符串)")
is_latest = Column(CHAR(1), default="0", comment="是否最新版本 1是 0否")
tenant_no = Column(String(64), nullable=False, default="zkxw", comment="租户编号")
order_num = Column(Integer, nullable=True, comment="排序号")
remark = Column(String(512), nullable=True, comment="备注")
create_by = Column(String(64), nullable=True, comment="创建人id")
create_dept_id = Column(String(64), nullable=False, default="", comment="创建人部门ID")
create_time = Column(DateTime, nullable=False, default=datetime.now, comment="创建时间")
update_by = Column(String(64), nullable=True, comment="更新人id")
update_time = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now, comment="更新时间")
del_flag = Column(SmallInteger, nullable=False, default=0, comment="删除标记 0正常 1已删除")
class WfNode(Base):
"""流程节点 wf_node"""
__tablename__ = "wf_node"
id = Column(String(64), primary_key=True, comment="主键ID")
process_version_id = Column(String(64), nullable=False, comment="流程版本ID")
node_key = Column(String(64), nullable=False, comment="节点唯一键(画布内)")
node_type = Column(String(32), nullable=False, comment="start/end/approval/condition等")
node_name = Column(String(200), nullable=True, comment="节点名称")
config_json = Column(Text, nullable=True, comment="审批人/SQL/会签策略等配置")
tenant_no = Column(String(64), nullable=False, default="zkxw", comment="租户编号")
order_num = Column(Integer, nullable=True, comment="排序号")
remark = Column(String(512), nullable=True, comment="备注")
create_by = Column(String(64), nullable=True, comment="创建人id")
create_dept_id = Column(String(64), nullable=False, default="", comment="创建人部门ID")
create_time = Column(DateTime, nullable=False, default=datetime.now, comment="创建时间")
update_by = Column(String(64), nullable=True, comment="更新人id")
update_time = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now, comment="更新时间")
del_flag = Column(SmallInteger, nullable=False, default=0, comment="删除标记 0正常 1已删除")
class WfTransition(Base):
"""流程连线/转移 wf_transition"""
__tablename__ = "wf_transition"
id = Column(String(64), primary_key=True, comment="主键ID")
process_version_id = Column(String(64), nullable=False, comment="流程版本ID")
from_node_key = Column(String(64), nullable=False, comment="起始节点key")
to_node_key = Column(String(64), nullable=False, comment="目标节点key")
condition_expr = Column(String(512), nullable=True, comment="条件表达式JSON")
tenant_no = Column(String(64), nullable=False, default="zkxw", comment="租户编号")
order_num = Column(Integer, nullable=True, comment="排序号")
remark = Column(String(512), nullable=True, comment="备注")
create_by = Column(String(64), nullable=True, comment="创建人id")
create_dept_id = Column(String(64), nullable=False, default="", comment="创建人部门ID")
create_time = Column(DateTime, nullable=False, default=datetime.now, comment="创建时间")
update_by = Column(String(64), nullable=True, comment="更新人id")
update_time = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now, comment="更新时间")
del_flag = Column(SmallInteger, nullable=False, default=0, comment="删除标记 0正常 1已删除")
class WfProcessInstance(Base):
"""流程实例 wf_process_instance"""
__tablename__ = "wf_process_instance"
id = Column(String(64), primary_key=True, comment="主键ID")
process_version_id = Column(String(64), nullable=False, comment="发起时锁定的流程版本ID")
business_type = Column(String(64), nullable=True, comment="业务类型")
business_id = Column(String(64), nullable=True, comment="业务主键")
starter_user_id = Column(String(64), nullable=True, comment="发起人")
variables_json = Column(Text, nullable=True, comment="流程变量JSON")
status = Column(String(32), nullable=True, comment="running/completed/rejected/cancelled")
callback_url = Column(String(512), nullable=True, comment="业务回调地址")
tenant_no = Column(String(64), nullable=False, default="zkxw", comment="租户编号")
order_num = Column(Integer, nullable=True, comment="排序号")
remark = Column(String(512), nullable=True, comment="备注")
create_by = Column(String(64), nullable=True, comment="创建人id")
create_dept_id = Column(String(64), nullable=False, default="", comment="创建人部门ID")
create_time = Column(DateTime, nullable=False, default=datetime.now, comment="创建时间")
update_by = Column(String(64), nullable=True, comment="更新人id")
update_time = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now, comment="更新时间")
del_flag = Column(SmallInteger, nullable=False, default=0, comment="删除标记 0正常 1已删除")
class WfTask(Base):
"""流程任务/待办 wf_task"""
__tablename__ = "wf_task"
id = Column(String(64), primary_key=True, comment="主键ID")
instance_id = Column(String(64), nullable=False, comment="流程实例ID")
node_key = Column(String(64), nullable=True, comment="当前节点key")
assignee_user_id = Column(String(64), nullable=True, comment="待办人")
status = Column(String(32), nullable=True, comment="pending/approved/rejected/withdrawn")
comment_text = Column(String(512), nullable=True, comment="审批意见")
tenant_no = Column(String(64), nullable=False, default="zkxw", comment="租户编号")
order_num = Column(Integer, nullable=True, comment="排序号")
remark = Column(String(512), nullable=True, comment="备注")
create_by = Column(String(64), nullable=True, comment="创建人id")
create_dept_id = Column(String(64), nullable=False, default="", comment="创建人部门ID")
create_time = Column(DateTime, nullable=False, default=datetime.now, comment="创建时间")
update_by = Column(String(64), nullable=True, comment="更新人id")
update_time = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now, comment="更新时间")
del_flag = Column(SmallInteger, nullable=False, default=0, comment="删除标记 0正常 1已删除")
- Step 2: 创建Schema变更SQL
在目标库执行下列 ALTER TABLE 后,将变更纳入 docs/sql/zkxw.sql(重导)或 docs/sql/ 下增量补丁:
-- 为 wf_process_instance 新增 callback_url 字段
ALTER TABLE `wf_process_instance`
ADD COLUMN `callback_url` VARCHAR(512) DEFAULT NULL COMMENT '业务回调地址' AFTER `status`;
- Step 3: 验证模型可导入
Run: cd backend && python -c "from app.models.models import WfProcessDef, WfProcessVersion, WfNode, WfTransition, WfProcessInstance, WfTask; print('OK')"
Expected: OK
Task 2: Pydantic Schemas
Files:
-
Create:
backend/app/schemas/workflow.py -
Step 1: 创建workflow schemas文件
"""Pydantic schemas - 工作流管理"""
from pydantic import BaseModel, Field
from typing import Optional, List, Any
# ---------------------------------------------------------------------------
# 流程定义
# ---------------------------------------------------------------------------
class ProcessDefCreate(BaseModel):
processCode: str = Field(..., min_length=1, max_length=64, description="流程编码")
processName: str = Field(..., min_length=1, max_length=200, description="流程名称")
status: Optional[str] = Field("1", description="1启用 0停用")
orderNum: Optional[int] = Field(0, description="排序号")
remark: Optional[str] = Field(None, description="备注")
class ProcessDefUpdate(BaseModel):
processCode: Optional[str] = Field(None, min_length=1, max_length=64)
processName: Optional[str] = Field(None, min_length=1, max_length=200)
status: Optional[str] = None
orderNum: Optional[int] = None
remark: Optional[str] = None
class ProcessDefResponse(BaseModel):
id: str
processCode: str
processName: str
status: str
latestVersion: Optional[int] = None
tenantNo: str
orderNum: int
remark: Optional[str] = None
createTime: str
updateTime: str
# ---------------------------------------------------------------------------
# 发布版本
# ---------------------------------------------------------------------------
class NodeData(BaseModel):
nodeKey: str = Field(..., description="节点唯一键")
nodeType: str = Field(..., description="start/end/approval/condition")
nodeName: Optional[str] = None
configJson: Optional[str] = None
class TransitionData(BaseModel):
fromNodeKey: str
toNodeKey: str
conditionExpr: Optional[str] = None
class PublishVersionRequest(BaseModel):
graphJson: str = Field(..., description="AntV X6画布JSON快照")
nodes: List[NodeData] = Field(default_factory=list)
transitions: List[TransitionData] = Field(default_factory=list)
remark: Optional[str] = None
class VersionResponse(BaseModel):
id: str
versionNo: int
isLatest: str
remark: Optional[str] = None
createTime: str
updateTime: str
# ---------------------------------------------------------------------------
# 发起流程
# ---------------------------------------------------------------------------
class StartInstanceRequest(BaseModel):
processCode: str = Field(..., description="流程编码")
businessType: Optional[str] = Field(None, description="业务类型")
businessId: Optional[str] = Field(None, description="业务主键")
variables: Optional[dict] = Field(default_factory=dict, description="流程变量")
callbackUrl: Optional[str] = Field(None, description="业务回调地址")
class InstanceResponse(BaseModel):
id: str
processVersionId: str
businessType: Optional[str] = None
businessId: Optional[str] = None
starterUserId: Optional[str] = None
status: Optional[str] = None
createTime: str
updateTime: str
tasks: Optional[List[dict]] = None
# ---------------------------------------------------------------------------
# 任务
# ---------------------------------------------------------------------------
class ApproveRequest(BaseModel):
commentText: Optional[str] = Field(None, description="审批意见")
class TaskResponse(BaseModel):
id: str
instanceId: str
nodeKey: Optional[str] = None
assigneeUserId: Optional[str] = None
status: Optional[str] = None
commentText: Optional[str] = None
createTime: str
updateTime: str
# 关联信息(列表页用)
processName: Optional[str] = None
businessType: Optional[str] = None
businessId: Optional[str] = None
starterUserId: Optional[str] = None
instanceStatus: Optional[str] = None
instanceCreateTime: Optional[str] = None
- Step 2: 验证schemas可导入
Run: cd backend && python -c "from app.schemas.workflow import ProcessDefCreate, StartInstanceRequest; print('OK')"
Expected: OK
Task 3: CRUD Operations
Files:
-
Create:
backend/app/crud/workflow.py -
Step 1: 创建workflow CRUD文件
遵循项目现有CRUD模式(独立函数、_row_to_dict、generate_uuid()、手动字段映射)。
"""工作流 CRUD 操作"""
from __future__ import annotations
import json
import math
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple
from sqlalchemy import and_, func, or_
from sqlalchemy.orm import Session
from app.core.utils import generate_uuid
from app.models.models import (
WfProcessDef,
WfProcessVersion,
WfNode,
WfTransition,
WfProcessInstance,
WfTask,
SysUser,
)
# ---------------------------------------------------------------------------
# 流程定义
# ---------------------------------------------------------------------------
def _def_to_dict(t: WfProcessDef, latest_version: Optional[int] = None) -> Dict[str, Any]:
return {
"id": t.id,
"processCode": t.process_code,
"processName": t.process_name,
"status": t.status,
"latestVersion": latest_version,
"tenantNo": t.tenant_no,
"orderNum": t.order_num if t.order_num is not None else 0,
"remark": t.remark or "",
"createTime": t.create_time.strftime("%Y-%m-%d %H:%M:%S") if t.create_time else "",
"updateTime": t.update_time.strftime("%Y-%m-%d %H:%M:%S") if t.update_time else "",
}
def list_process_defs(
db: Session,
*,
page: int = 1,
page_size: int = 10,
keyword: Optional[str] = None,
status: Optional[str] = None,
tenant_no: Optional[str] = None,
) -> Tuple[List[Dict[str, Any]], int]:
def _filtered():
q0 = db.query(WfProcessDef).filter(WfProcessDef.del_flag == 0)
if tenant_no:
q0 = q0.filter(WfProcessDef.tenant_no == tenant_no)
if status:
q0 = q0.filter(WfProcessDef.status == status)
kw = (keyword or "").strip()
if kw:
like = f"%{kw}%"
q0 = q0.filter(
or_(WfProcessDef.process_name.like(like), WfProcessDef.process_code.like(like))
)
return q0
total = _filtered().order_by(None).with_entities(func.count(WfProcessDef.id)).scalar() or 0
page = max(1, page)
page_size = min(max(1, page_size), 100)
rows = (
_filtered()
.order_by(WfProcessDef.order_num.asc(), WfProcessDef.create_time.desc())
.offset((page - 1) * page_size)
.limit(page_size)
.all()
)
result = []
for r in rows:
latest = (
db.query(func.max(WfProcessVersion.version_no))
.filter(
WfProcessVersion.process_def_id == r.id,
WfProcessVersion.del_flag == 0,
)
.scalar()
)
result.append(_def_to_dict(r, latest_version=latest))
return result, int(total)
def get_process_def(db: Session, def_id: str) -> Optional[WfProcessDef]:
return (
db.query(WfProcessDef)
.filter(and_(WfProcessDef.id == def_id, WfProcessDef.del_flag == 0))
.first()
)
def process_code_exists(
db: Session, code: str, tenant_no: str, exclude_id: Optional[str] = None
) -> bool:
q = db.query(WfProcessDef).filter(
and_(WfProcessDef.process_code == code.strip(), WfProcessDef.tenant_no == tenant_no, WfProcessDef.del_flag == 0)
)
if exclude_id:
q = q.filter(WfProcessDef.id != exclude_id)
return q.first() is not None
def create_process_def(
db: Session,
*,
process_code: str,
process_name: str,
tenant_no: str,
status: str = "1",
order_num: int = 0,
remark: Optional[str] = None,
user_id: Optional[str] = None,
) -> WfProcessDef:
code = (process_code or "").strip()
if not code:
raise ValueError("流程编码不能为空")
if process_code_exists(db, code, tenant_no):
raise ValueError("流程编码已存在")
obj = WfProcessDef(
id=generate_uuid(),
process_code=code,
process_name=(process_name or "").strip(),
status=status,
tenant_no=tenant_no,
order_num=order_num,
remark=(remark or "").strip() or None,
create_by=user_id,
create_dept_id="",
create_time=datetime.now(),
update_by=user_id,
update_time=datetime.now(),
del_flag=0,
)
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def update_process_def(
db: Session,
def_id: str,
*,
process_code: Optional[str] = None,
process_name: Optional[str] = None,
status: Optional[str] = None,
order_num: Optional[int] = None,
remark: Optional[str] = None,
user_id: Optional[str] = None,
tenant_no: Optional[str] = None,
) -> WfProcessDef:
obj = get_process_def(db, def_id)
if not obj:
raise ValueError("流程定义不存在")
if process_code is not None:
new_code = process_code.strip()
if not new_code:
raise ValueError("流程编码不能为空")
tn = tenant_no or obj.tenant_no
if new_code != obj.process_code and process_code_exists(db, new_code, tn, exclude_id=def_id):
raise ValueError("流程编码已存在")
obj.process_code = new_code
if process_name is not None:
obj.process_name = (process_name or "").strip()
if status is not None:
obj.status = status
if order_num is not None:
obj.order_num = order_num
if remark is not None:
obj.remark = (remark or "").strip() or None
obj.update_by = user_id
obj.update_time = datetime.now()
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def soft_delete_process_def(db: Session, def_id: str) -> None:
obj = get_process_def(db, def_id)
if not obj:
raise ValueError("流程定义不存在")
obj.del_flag = 1
obj.update_time = datetime.now()
db.add(obj)
db.commit()
# ---------------------------------------------------------------------------
# 发布版本
# ---------------------------------------------------------------------------
def publish_version(
db: Session,
*,
process_def_id: str,
graph_json: str,
nodes: List[Dict[str, Any]],
transitions: List[Dict[str, Any]],
tenant_no: str,
remark: Optional[str] = None,
user_id: Optional[str] = None,
) -> WfProcessVersion:
# 将旧版本 is_latest 置 0
db.query(WfProcessVersion).filter(
WfProcessVersion.process_def_id == process_def_id,
WfProcessVersion.is_latest == "1",
WfProcessVersion.del_flag == 0,
).update({"is_latest": "0"}, synchronize_session=False)
# 计算新版本号
max_ver = (
db.query(func.max(WfProcessVersion.version_no))
.filter(WfProcessVersion.process_def_id == process_def_id, WfProcessVersion.del_flag == 0)
.scalar()
) or 0
new_ver = max_ver + 1
# 创建新版本
version_id = generate_uuid()
version = WfProcessVersion(
id=version_id,
process_def_id=process_def_id,
version_no=new_ver,
graph_json=graph_json,
is_latest="1",
tenant_no=tenant_no,
remark=(remark or "").strip() or None,
create_by=user_id,
create_dept_id="",
create_time=datetime.now(),
update_by=user_id,
update_time=datetime.now(),
del_flag=0,
)
db.add(version)
# 写入节点
for n in nodes:
node = WfNode(
id=generate_uuid(),
process_version_id=version_id,
node_key=n.get("nodeKey", ""),
node_type=n.get("nodeType", ""),
node_name=n.get("nodeName"),
config_json=n.get("configJson"),
tenant_no=tenant_no,
create_by=user_id,
create_dept_id="",
create_time=datetime.now(),
update_by=user_id,
update_time=datetime.now(),
del_flag=0,
)
db.add(node)
# 写入连线
for t in transitions:
trans = WfTransition(
id=generate_uuid(),
process_version_id=version_id,
from_node_key=t.get("fromNodeKey", ""),
to_node_key=t.get("toNodeKey", ""),
condition_expr=t.get("conditionExpr"),
tenant_no=tenant_no,
create_by=user_id,
create_dept_id="",
create_time=datetime.now(),
update_by=user_id,
update_time=datetime.now(),
del_flag=0,
)
db.add(trans)
db.commit()
db.refresh(version)
return version
def get_latest_version(db: Session, process_def_id: str) -> Optional[WfProcessVersion]:
return (
db.query(WfProcessVersion)
.filter(
WfProcessVersion.process_def_id == process_def_id,
WfProcessVersion.is_latest == "1",
WfProcessVersion.del_flag == 0,
)
.first()
)
def list_versions(
db: Session, process_def_id: str, tenant_no: Optional[str] = None
) -> List[Dict[str, Any]]:
q = db.query(WfProcessVersion).filter(
WfProcessVersion.process_def_id == process_def_id,
WfProcessVersion.del_flag == 0,
)
if tenant_no:
q = q.filter(WfProcessVersion.tenant_no == tenant_no)
rows = q.order_by(WfProcessVersion.version_no.desc()).all()
return [
{
"id": v.id,
"versionNo": v.version_no,
"isLatest": v.is_latest,
"remark": v.remark or "",
"createTime": v.create_time.strftime("%Y-%m-%d %H:%M:%S") if v.create_time else "",
"updateTime": v.update_time.strftime("%Y-%m-%d %H:%M:%S") if v.update_time else "",
}
for v in rows
]
# ---------------------------------------------------------------------------
# 流程实例
# ---------------------------------------------------------------------------
def create_instance(
db: Session,
*,
process_version_id: str,
business_type: Optional[str] = None,
business_id: Optional[str] = None,
starter_user_id: Optional[str] = None,
variables: Optional[dict] = None,
callback_url: Optional[str] = None,
tenant_no: str = "zkxw",
user_id: Optional[str] = None,
) -> WfProcessInstance:
obj = WfProcessInstance(
id=generate_uuid(),
process_version_id=process_version_id,
business_type=business_type,
business_id=business_id,
starter_user_id=starter_user_id,
variables_json=json.dumps(variables or {}, ensure_ascii=False),
status="running",
callback_url=callback_url,
tenant_no=tenant_no,
create_by=user_id,
create_dept_id="",
create_time=datetime.now(),
update_by=user_id,
update_time=datetime.now(),
del_flag=0,
)
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def get_instance(db: Session, instance_id: str) -> Optional[WfProcessInstance]:
return (
db.query(WfProcessInstance)
.filter(and_(WfProcessInstance.id == instance_id, WfProcessInstance.del_flag == 0))
.first()
)
def update_instance_status(db: Session, instance_id: str, status: str) -> None:
db.query(WfProcessInstance).filter(WfProcessInstance.id == instance_id).update(
{"status": status, "update_time": datetime.now()}, synchronize_session=False
)
db.commit()
# ---------------------------------------------------------------------------
# 任务
# ---------------------------------------------------------------------------
def create_task(
db: Session,
*,
instance_id: str,
node_key: str,
assignee_user_id: str,
tenant_no: str = "zkxw",
user_id: Optional[str] = None,
) -> WfTask:
obj = WfTask(
id=generate_uuid(),
instance_id=instance_id,
node_key=node_key,
assignee_user_id=assignee_user_id,
status="pending",
tenant_no=tenant_no,
create_by=user_id,
create_dept_id="",
create_time=datetime.now(),
update_by=user_id,
update_time=datetime.now(),
del_flag=0,
)
db.add(obj)
db.commit()
db.refresh(obj)
return obj
def get_task(db: Session, task_id: str) -> Optional[WfTask]:
return (
db.query(WfTask)
.filter(and_(WfTask.id == task_id, WfTask.del_flag == 0))
.first()
)
def get_pending_tasks_by_node(
db: Session, instance_id: str, node_key: str
) -> List[WfTask]:
return (
db.query(WfTask)
.filter(
WfTask.instance_id == instance_id,
WfTask.node_key == node_key,
WfTask.status == "pending",
WfTask.del_flag == 0,
)
.all()
)
def list_pending_tasks(
db: Session,
*,
user_id: str,
page: int = 1,
page_size: int = 10,
keyword: Optional[str] = None,
tenant_no: Optional[str] = None,
) -> Tuple[List[Dict[str, Any]], int]:
"""我的待办列表"""
q = (
db.query(WfTask, WfProcessInstance, WfProcessVersion, WfProcessDef)
.join(WfProcessInstance, WfTask.instance_id == WfProcessInstance.id)
.join(WfProcessVersion, WfProcessInstance.process_version_id == WfProcessVersion.id)
.join(WfProcessDef, WfProcessVersion.process_def_id == WfProcessDef.id)
.filter(
WfTask.assignee_user_id == user_id,
WfTask.status == "pending",
WfTask.del_flag == 0,
WfProcessInstance.del_flag == 0,
)
)
if tenant_no:
q = q.filter(WfTask.tenant_no == tenant_no)
if keyword and keyword.strip():
like = f"%{keyword.strip()}%"
q = q.filter(WfProcessDef.process_name.like(like))
total = q.count()
page = max(1, page)
page_size = min(max(1, page_size), 100)
rows = (
q.order_by(WfTask.create_time.desc())
.offset((page - 1) * page_size)
.limit(page_size)
.all()
)
result = []
for task, inst, ver, pdef in rows:
result.append({
"id": task.id,
"instanceId": task.instance_id,
"nodeKey": task.node_key,
"assigneeUserId": task.assignee_user_id,
"status": task.status,
"createTime": task.create_time.strftime("%Y-%m-%d %H:%M:%S") if task.create_time else "",
"updateTime": task.update_time.strftime("%Y-%m-%d %H:%M:%S") if task.update_time else "",
"processName": pdef.process_name,
"businessType": inst.business_type,
"businessId": inst.business_id,
"starterUserId": inst.starter_user_id,
"instanceStatus": inst.status,
"instanceCreateTime": inst.create_time.strftime("%Y-%m-%d %H:%M:%S") if inst.create_time else "",
})
return result, int(total)
def list_completed_tasks(
db: Session,
*,
user_id: str,
page: int = 1,
page_size: int = 10,
keyword: Optional[str] = None,
tenant_no: Optional[str] = None,
) -> Tuple[List[Dict[str, Any]], int]:
"""我的已办列表"""
q = (
db.query(WfTask, WfProcessInstance, WfProcessVersion, WfProcessDef)
.join(WfProcessInstance, WfTask.instance_id == WfProcessInstance.id)
.join(WfProcessVersion, WfProcessInstance.process_version_id == WfProcessVersion.id)
.join(WfProcessDef, WfProcessVersion.process_def_id == WfProcessDef.id)
.filter(
WfTask.assignee_user_id == user_id,
WfTask.status.in_(["approved", "rejected"]),
WfTask.del_flag == 0,
WfProcessInstance.del_flag == 0,
)
)
if tenant_no:
q = q.filter(WfTask.tenant_no == tenant_no)
if keyword and keyword.strip():
like = f"%{keyword.strip()}%"
q = q.filter(WfProcessDef.process_name.like(like))
total = q.count()
page = max(1, page)
page_size = min(max(1, page_size), 100)
rows = (
q.order_by(WfTask.update_time.desc())
.offset((page - 1) * page_size)
.limit(page_size)
.all()
)
result = []
for task, inst, ver, pdef in rows:
result.append({
"id": task.id,
"instanceId": task.instance_id,
"nodeKey": task.node_key,
"assigneeUserId": task.assignee_user_id,
"status": task.status,
"commentText": task.comment_text,
"createTime": task.create_time.strftime("%Y-%m-%d %H:%M:%S") if task.create_time else "",
"updateTime": task.update_time.strftime("%Y-%m-%d %H:%M:%S") if task.update_time else "",
"processName": pdef.process_name,
"businessType": inst.business_type,
"businessId": inst.business_id,
"starterUserId": inst.starter_user_id,
"instanceStatus": inst.status,
"instanceCreateTime": inst.create_time.strftime("%Y-%m-%d %H:%M:%S") if inst.create_time else "",
})
return result, int(total)
# ---------------------------------------------------------------------------
# 节点/连线查询
# ---------------------------------------------------------------------------
def get_nodes_by_version(db: Session, version_id: str) -> List[WfNode]:
return (
db.query(WfNode)
.filter(WfNode.process_version_id == version_id, WfNode.del_flag == 0)
.all()
)
def get_transitions_by_version(db: Session, version_id: str) -> List[WfTransition]:
return (
db.query(WfTransition)
.filter(WfTransition.process_version_id == version_id, WfTransition.del_flag == 0)
.all()
)
def total_pages(total: int, page_size: int) -> int:
if total <= 0:
return 0
return int(math.ceil(total / float(page_size)))
- Step 2: 验证CRUD可导入
Run: cd backend && python -c "from app.crud.workflow import list_process_defs, publish_version; print('OK')"
Expected: OK
Phase 2: Workflow Engine
Task 4: Core Engine(节点分发 + Start/End Handler)
Files:
-
Create:
backend/app/services/workflow_engine.py -
Step 1: 创建工作流引擎
"""工作流引擎 - 图遍历模式"""
from __future__ import annotations
import json
import logging
import re
from datetime import datetime
from typing import Any, Dict, List, Optional
from sqlalchemy.orm import Session
from app.crud import workflow as wf_crud
from app.models.models import WfNode, WfTransition, WfTask
logger = logging.getLogger(__name__)
class WorkflowEngine:
"""图遍历工作流引擎"""
def __init__(self, db: Session):
self.db = db
# ------------------------------------------------------------------
# 公共入口
# ------------------------------------------------------------------
def start_process(
self,
*,
process_version_id: str,
business_type: Optional[str] = None,
business_id: Optional[str] = None,
starter_user_id: Optional[str] = None,
variables: Optional[dict] = None,
callback_url: Optional[str] = None,
tenant_no: str = "zkxw",
user_id: Optional[str] = None,
) -> str:
"""发起流程,返回实例ID"""
instance = wf_crud.create_instance(
self.db,
process_version_id=process_version_id,
business_type=business_type,
business_id=business_id,
starter_user_id=starter_user_id,
variables=variables,
callback_url=callback_url,
tenant_no=tenant_no,
user_id=user_id,
)
# 找到 start 节点并推进
nodes = wf_crud.get_nodes_by_version(self.db, process_version_id)
start_node = next((n for n in nodes if n.node_type == "start"), None)
if start_node:
self._advance(instance.id, start_node.node_key, tenant_no)
return instance.id
def approve_task(self, task_id: str, comment: Optional[str] = None, tenant_no: str = "zkxw") -> None:
"""审批通过任务"""
task = wf_crud.get_task(self.db, task_id)
if not task or task.status != "pending":
raise ValueError("任务不存在或已处理")
task.status = "approved"
task.comment_text = comment
task.update_time = datetime.now()
self.db.add(task)
self.db.commit()
# 检查同节点是否全部通过
instance = wf_crud.get_instance(self.db, task.instance_id)
if not instance:
return
nodes = wf_crud.get_nodes_by_version(self.db, instance.process_version_id)
current_node = next((n for n in nodes if n.node_key == task.node_key), None)
if not current_node:
return
config = json.loads(current_node.config_json or "{}")
pass_type = config.get("pass_type", "any")
pending_tasks = wf_crud.get_pending_tasks_by_node(self.db, task.instance_id, task.node_key)
should_advance = False
if pass_type == "any":
# 任一通过即可推进,取消其余待办
should_advance = True
for pt in pending_tasks:
pt.status = "withdrawn"
pt.update_time = datetime.now()
self.db.add(pt)
self.db.commit()
else:
# all: 全部通过才推进
if not pending_tasks:
should_advance = True
if should_advance:
self._advance(task.instance_id, task.node_key, tenant_no)
def reject_task(self, task_id: str, comment: Optional[str] = None, tenant_no: str = "zkxw") -> None:
"""审批驳回任务"""
task = wf_crud.get_task(self.db, task_id)
if not task or task.status != "pending":
raise ValueError("任务不存在或已处理")
task.status = "rejected"
task.comment_text = comment
task.update_time = datetime.now()
self.db.add(task)
self.db.commit()
# 驳回 → 取消同节点其他待办 → 实例标记 rejected
pending_tasks = wf_crud.get_pending_tasks_by_node(self.db, task.instance_id, task.node_key)
for pt in pending_tasks:
pt.status = "withdrawn"
pt.update_time = datetime.now()
self.db.add(pt)
self.db.commit()
wf_crud.update_instance_status(self.db, task.instance_id, "rejected")
# 触发回调
instance = wf_crud.get_instance(self.db, task.instance_id)
if instance:
from app.services.workflow_callback import trigger_callback
trigger_callback(self.db, instance, event="rejected")
# ------------------------------------------------------------------
# 内部推进
# ------------------------------------------------------------------
def _advance(self, instance_id: str, from_node_key: str, tenant_no: str) -> None:
"""从指定节点推进到下一个节点"""
instance = wf_crud.get_instance(self.db, instance_id)
if not instance or instance.status != "running":
return
nodes = wf_crud.get_nodes_by_version(self.db, instance.process_version_id)
transitions = wf_crud.get_transitions_by_version(self.db, instance.process_version_id)
variables = json.loads(instance.variables_json or "{}")
current_node = next((n for n in nodes if n.node_key == from_node_key), None)
if not current_node:
return
# 找到出边
out_edges = [t for t in transitions if t.from_node_key == from_node_key]
if not out_edges:
return
# 根据节点类型决定下一条边
next_node_key = None
if current_node.node_type == "start":
next_node_key = out_edges[0].to_node_key
elif current_node.node_type == "approval":
next_node_key = out_edges[0].to_node_key if out_edges else None
elif current_node.node_type == "condition":
next_node_key = self._evaluate_condition(out_edges, variables)
else:
return
if not next_node_key:
return
# 执行目标节点
next_node = next((n for n in nodes if n.node_key == next_node_key), None)
if not next_node:
return
self._execute_node(instance_id, next_node, variables, tenant_no)
def _execute_node(self, instance_id: str, node: WfNode, variables: dict, tenant_no: str) -> None:
"""执行节点逻辑"""
if node.node_type == "end":
wf_crud.update_instance_status(self.db, instance_id, "completed")
instance = wf_crud.get_instance(self.db, instance_id)
if instance:
from app.services.workflow_callback import trigger_callback
trigger_callback(self.db, instance, event="completed")
elif node.node_type == "approval":
self._create_approval_tasks(instance_id, node, variables, tenant_no)
elif node.node_type == "condition":
self._advance(instance_id, node.node_key, tenant_no)
else:
# start 等其他类型直接推进
self._advance(instance_id, node.node_key, tenant_no)
def _create_approval_tasks(
self, instance_id: str, node: WfNode, variables: dict, tenant_no: str
) -> None:
"""审批节点:解析审批人并创建任务"""
config = json.loads(node.config_json or "{}")
user_ids = self._resolve_approvers(config, variables, tenant_no)
for uid in user_ids:
wf_crud.create_task(
self.db,
instance_id=instance_id,
node_key=node.node_key,
assignee_user_id=uid,
tenant_no=tenant_no,
)
def _resolve_approvers(self, config: dict, variables: dict, tenant_no: str) -> List[str]:
"""解析审批人列表"""
strategy = config.get("strategy", "users")
if strategy == "users":
return config.get("user_ids", [])
elif strategy == "roles":
role_ids = config.get("role_ids", [])
if not role_ids:
return []
from app.models.models import SysUserRole, SysUser
rows = (
self.db.query(SysUser.id)
.join(SysUserRole, SysUser.id == SysUserRole.user_id)
.filter(
SysUserRole.role_id.in_(role_ids),
SysUser.del_flag == 0,
SysUser.status == "1",
)
.distinct()
.all()
)
return [r[0] for r in rows]
elif strategy == "sql":
sql_template = config.get("sql", "")
if not sql_template:
return []
return self._execute_sql_approvers(sql_template, variables, tenant_no)
return []
def _execute_sql_approvers(self, sql_template: str, variables: dict, tenant_no: str) -> List[str]:
"""执行SQL解析审批人(仅允许SELECT,参数化查询)"""
# 安全检查:只允许 SELECT
normalized = sql_template.strip().upper()
for forbidden in ["INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "CREATE", "TRUNCATE"]:
if forbidden in normalized:
logger.warning(f"SQL approver resolution blocked: forbidden keyword '{forbidden}'")
return []
if not normalized.startswith("SELECT"):
return []
# 参数替换
params = dict(variables)
params["tenant_no"] = tenant_no
try:
result = self.db.execute(sql_template, params)
rows = result.fetchall()
return [str(r[0]) for r in rows]
except Exception as e:
logger.error(f"SQL approver resolution failed: {e}")
return []
def _evaluate_condition(self, out_edges: List[WfTransition], variables: dict) -> Optional[str]:
"""评估条件分支,返回目标节点key"""
for edge in out_edges:
if not edge.condition_expr:
# 无条件 = 默认分支(最后一条)
continue
try:
rules = json.loads(edge.condition_expr)
if self._check_rules(rules, variables):
return edge.to_node_key
except (json.JSONDecodeError, KeyError):
continue
# 兜底:最后一条无条件边
for edge in reversed(out_edges):
if not edge.condition_expr:
return edge.to_node_key
return out_edges[-1].to_node_key if out_edges else None
def _check_rules(self, rules: Any, variables: dict) -> bool:
"""检查条件规则"""
if isinstance(rules, dict):
if "rules" in rules:
# 顶层 rules 数组
return self._check_rules(rules["rules"], variables)
if "and" in rules:
return all(self._check_rules(r, variables) for r in rules["and"])
if "or" in rules:
return any(self._check_rules(r, variables) for r in rules["or"])
# 单条规则
field = rules.get("field", "")
op = rules.get("op", "==")
value = rules.get("value")
actual = self._get_variable(field, variables)
return self._compare(actual, op, value)
elif isinstance(rules, list):
# 如果是数组,第一条匹配即返回(按顺序评估)
for rule in rules:
if self._check_rules(rule, variables):
return True
return False
return False
def _get_variable(self, field: str, variables: dict) -> Any:
"""从变量字典取值,支持点号路径"""
parts = field.split(".")
val = variables
for p in parts:
if isinstance(val, dict):
val = val.get(p)
else:
return None
return val
def _compare(self, actual: Any, op: str, expected: Any) -> bool:
"""比较操作"""
try:
if op == "==":
return actual == expected
elif op == "!=":
return actual != expected
elif op == ">":
return float(actual) > float(expected)
elif op == "<":
return float(actual) < float(expected)
elif op == ">=":
return float(actual) >= float(expected)
elif op == "<=":
return float(actual) <= float(expected)
elif op == "in":
return actual in expected if isinstance(expected, list) else False
elif op == "not_in":
return actual not in expected if isinstance(expected, list) else True
elif op == "contains":
return str(expected) in str(actual) if actual else False
except (TypeError, ValueError):
return False
return False
- Step 2: 验证引擎可导入
Run: cd backend && python -c "from app.services.workflow_engine import WorkflowEngine; print('OK')"
Expected: OK
Task 5: Callback Service
Files:
-
Create:
backend/app/services/workflow_callback.py -
Step 1: 创建回调服务
"""工作流回调服务"""
from __future__ import annotations
import json
import logging
from typing import Optional
import requests
from sqlalchemy.orm import Session
from app.models.models import WfProcessInstance
logger = logging.getLogger(__name__)
CALLBACK_TIMEOUT_SECONDS = 10
def trigger_callback(db: Session, instance: WfProcessInstance, event: str) -> None:
"""触发业务回调"""
if not instance.callback_url:
return
payload = {
"event": event,
"instanceId": instance.id,
"businessType": instance.business_type,
"businessId": instance.business_id,
"status": instance.status,
}
try:
resp = requests.post(
instance.callback_url,
json=payload,
timeout=CALLBACK_TIMEOUT_SECONDS,
headers={"Content-Type": "application/json"},
)
logger.info(
f"Workflow callback triggered: {instance.callback_url} event={event} status={resp.status_code}"
)
except Exception as e:
logger.error(f"Workflow callback failed: {instance.callback_url} error={e}")
- Step 2: 添加 requests 到 requirements.txt(如果缺失)
Run: cd backend && grep -q "requests" requirements.txt && echo "EXISTS" || echo "requests" >> requirements.txt
Phase 3: API Routes
Task 6: Workflow API Routes + Router Registration
Files:
-
Create:
backend/app/api/workflow.py -
Modify:
backend/app/api/__init__.py -
Modify:
backend/app/main.py -
Step 1: 创建workflow API路由文件
"""工作流管理 API 路由"""
from typing import Optional
from fastapi import APIRouter, Depends, Query
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy.orm import Session
from app.core.database import get_db
from app.core.menu_guard import require_admin_or_super_admin
from app.core.response import success_response, error_response
from app.crud import workflow as wf_crud
from app.schemas.workflow import (
ProcessDefCreate,
ProcessDefUpdate,
PublishVersionRequest,
StartInstanceRequest,
ApproveRequest,
)
from app.services.workflow_engine import WorkflowEngine
router = APIRouter(prefix="/web/admin/workflow", tags=["工作流管理"])
# ---------------------------------------------------------------------------
# 流程定义
# ---------------------------------------------------------------------------
@router.get("/defs")
async def list_defs(
page: int = Query(1, ge=1),
pageSize: int = Query(10, ge=1, le=100),
keyword: Optional[str] = None,
status: Optional[str] = None,
_user: dict = Depends(require_admin_or_super_admin),
db: Session = Depends(get_db),
):
tenant_no = _user.get("tenant_no", "zkxw")
rows, total = wf_crud.list_process_defs(
db, page=page, page_size=pageSize, keyword=keyword, status=status, tenant_no=tenant_no
)
return success_response({
"list": rows,
"total": total,
"page": page,
"pageSize": pageSize,
"totalPages": wf_crud.total_pages(total, pageSize),
})
class _DefBody(BaseModel):
model_config = ConfigDict(populate_by_name=True)
processCode: str = Field(..., alias="processCode")
processName: str = Field(..., alias="processName")
status: Optional[str] = Field("1")
orderNum: Optional[int] = Field(0)
remark: Optional[str] = None
@router.post("/defs")
async def create_def(
body: _DefBody,
_user: dict = Depends(require_admin_or_super_admin),
db: Session = Depends(get_db),
):
tenant_no = _user.get("tenant_no", "zkxw")
user_id = _user.get("user_id") or _user.get("id")
try:
obj = wf_crud.create_process_def(
db,
process_code=body.processCode,
process_name=body.processName,
tenant_no=tenant_no,
status=body.status or "1",
order_num=body.orderNum or 0,
remark=body.remark,
user_id=user_id,
)
return success_response(wf_crud._def_to_dict(obj))
except ValueError as e:
return error_response(str(e))
class _DefUpdateBody(BaseModel):
model_config = ConfigDict(populate_by_name=True)
processCode: Optional[str] = None
processName: Optional[str] = None
status: Optional[str] = None
orderNum: Optional[int] = None
remark: Optional[str] = None
@router.put("/defs/{def_id}")
async def update_def(
def_id: str,
body: _DefUpdateBody,
_user: dict = Depends(require_admin_or_super_admin),
db: Session = Depends(get_db),
):
tenant_no = _user.get("tenant_no", "zkxw")
user_id = _user.get("user_id") or _user.get("id")
try:
obj = wf_crud.update_process_def(
db,
def_id,
process_code=body.processCode,
process_name=body.processName,
status=body.status,
order_num=body.orderNum,
remark=body.remark,
user_id=user_id,
tenant_no=tenant_no,
)
return success_response(wf_crud._def_to_dict(obj))
except ValueError as e:
return error_response(str(e))
@router.delete("/defs/{def_id}")
async def delete_def(
def_id: str,
_user: dict = Depends(require_admin_or_super_admin),
db: Session = Depends(get_db),
):
try:
wf_crud.soft_delete_process_def(db, def_id)
return success_response(None, msg="删除成功")
except ValueError as e:
return error_response(str(e))
# ---------------------------------------------------------------------------
# 发布版本
# ---------------------------------------------------------------------------
@router.post("/defs/{def_id}/publish")
async def publish_version(
def_id: str,
body: PublishVersionRequest,
_user: dict = Depends(require_admin_or_super_admin),
db: Session = Depends(get_db),
):
tenant_no = _user.get("tenant_no", "zkxw")
user_id = _user.get("user_id") or _user.get("id")
def_obj = wf_crud.get_process_def(db, def_id)
if not def_obj:
return error_response("流程定义不存在")
nodes = [n.model_dump() for n in body.nodes]
transitions = [t.model_dump() for t in body.transitions]
version = wf_crud.publish_version(
db,
process_def_id=def_id,
graph_json=body.graphJson,
nodes=nodes,
transitions=transitions,
tenant_no=tenant_no,
remark=body.remark,
user_id=user_id,
)
return success_response({
"id": version.id,
"versionNo": version.version_no,
"isLatest": version.is_latest,
})
@router.get("/defs/{def_id}/versions")
async def list_versions(
def_id: str,
_user: dict = Depends(require_admin_or_super_admin),
db: Session = Depends(get_db),
):
tenant_no = _user.get("tenant_no", "zkxw")
rows = wf_crud.list_versions(db, def_id, tenant_no=tenant_no)
return success_response(rows)
# ---------------------------------------------------------------------------
# 获取设计器数据(最新版本的 graph_json + nodes + transitions)
# ---------------------------------------------------------------------------
@router.get("/defs/{def_id}/designer")
async def get_designer_data(
def_id: str,
_user: dict = Depends(require_admin_or_super_admin),
db: Session = Depends(get_db),
):
version = wf_crud.get_latest_version(db, def_id)
if not version:
return success_response({"graphJson": None, "nodes": [], "transitions": []})
nodes = wf_crud.get_nodes_by_version(db, version.id)
transitions = wf_crud.get_transitions_by_version(db, version.id)
return success_response({
"graphJson": version.graph_json,
"versionNo": version.version_no,
"nodes": [
{
"nodeKey": n.node_key,
"nodeType": n.node_type,
"nodeName": n.node_name,
"configJson": n.config_json,
}
for n in nodes
],
"transitions": [
{
"fromNodeKey": t.from_node_key,
"toNodeKey": t.to_node_key,
"conditionExpr": t.condition_expr,
}
for t in transitions
],
})
# ---------------------------------------------------------------------------
# 流程实例
# ---------------------------------------------------------------------------
@router.post("/instances")
async def start_instance(
body: StartInstanceRequest,
_user: dict = Depends(require_admin_or_super_admin),
db: Session = Depends(get_db),
):
tenant_no = _user.get("tenant_no", "zkxw")
user_id = _user.get("user_id") or _user.get("id")
# 查找流程定义
all_defs = db.query(wf_crud.WfProcessDef).filter(
wf_crud.WfProcessDef.process_code == body.processCode,
wf_crud.WfProcessDef.del_flag == 0,
wf_crud.WfProcessDef.tenant_no == tenant_no,
).first()
if not all_defs:
return error_response("流程定义不存在")
if all_defs.status != "1":
return error_response("流程已停用")
# 查找最新版本
version = wf_crud.get_latest_version(db, all_defs.id)
if not version:
return error_response("流程尚未发布版本")
engine = WorkflowEngine(db)
instance_id = engine.start_process(
process_version_id=version.id,
business_type=body.businessType,
business_id=body.businessId,
starter_user_id=user_id,
variables=body.variables,
callback_url=body.callbackUrl,
tenant_no=tenant_no,
user_id=user_id,
)
return success_response({"instanceId": instance_id})
@router.get("/instances/{instance_id}")
async def get_instance_detail(
instance_id: str,
_user: dict = Depends(require_admin_or_super_admin),
db: Session = Depends(get_db),
):
instance = wf_crud.get_instance(db, instance_id)
if not instance:
return error_response("流程实例不存在")
tasks = (
db.query(wf_crud.WfTask)
.filter(wf_crud.WfTask.instance_id == instance_id, wf_crud.WfTask.del_flag == 0)
.order_by(wf_crud.WfTask.create_time.asc())
.all()
)
return success_response({
"id": instance.id,
"processVersionId": instance.process_version_id,
"businessType": instance.business_type,
"businessId": instance.business_id,
"starterUserId": instance.starter_user_id,
"status": instance.status,
"variables": instance.variables_json,
"createTime": instance.create_time.strftime("%Y-%m-%d %H:%M:%S") if instance.create_time else "",
"tasks": [
{
"id": t.id,
"nodeKey": t.node_key,
"assigneeUserId": t.assignee_user_id,
"status": t.status,
"commentText": t.comment_text,
"createTime": t.create_time.strftime("%Y-%m-%d %H:%M:%S") if t.create_time else "",
"updateTime": t.update_time.strftime("%Y-%m-%d %H:%M:%S") if t.update_time else "",
}
for t in tasks
],
})
# ---------------------------------------------------------------------------
# 任务办理
# ---------------------------------------------------------------------------
@router.get("/tasks/pending")
async def list_pending(
page: int = Query(1, ge=1),
pageSize: int = Query(10, ge=1, le=100),
keyword: Optional[str] = None,
_user: dict = Depends(require_admin_or_super_admin),
db: Session = Depends(get_db),
):
user_id = _user.get("user_id") or _user.get("id")
tenant_no = _user.get("tenant_no", "zkxw")
rows, total = wf_crud.list_pending_tasks(
db, user_id=user_id, page=page, page_size=pageSize, keyword=keyword, tenant_no=tenant_no
)
return success_response({
"list": rows,
"total": total,
"page": page,
"pageSize": pageSize,
"totalPages": wf_crud.total_pages(total, pageSize),
})
@router.get("/tasks/completed")
async def list_completed(
page: int = Query(1, ge=1),
pageSize: int = Query(10, ge=1, le=100),
keyword: Optional[str] = None,
_user: dict = Depends(require_admin_or_super_admin),
db: Session = Depends(get_db),
):
user_id = _user.get("user_id") or _user.get("id")
tenant_no = _user.get("tenant_no", "zkxw")
rows, total = wf_crud.list_completed_tasks(
db, user_id=user_id, page=page, page_size=pageSize, keyword=keyword, tenant_no=tenant_no
)
return success_response({
"list": rows,
"total": total,
"page": page,
"pageSize": pageSize,
"totalPages": wf_crud.total_pages(total, pageSize),
})
@router.post("/tasks/{task_id}/approve")
async def approve_task(
task_id: str,
body: ApproveRequest,
_user: dict = Depends(require_admin_or_super_admin),
db: Session = Depends(get_db),
):
tenant_no = _user.get("tenant_no", "zkxw")
try:
engine = WorkflowEngine(db)
engine.approve_task(task_id, comment=body.commentText, tenant_no=tenant_no)
return success_response(None, msg="审批通过")
except ValueError as e:
return error_response(str(e))
@router.post("/tasks/{task_id}/reject")
async def reject_task(
task_id: str,
body: ApproveRequest,
_user: dict = Depends(require_admin_or_super_admin),
db: Session = Depends(get_db),
):
tenant_no = _user.get("tenant_no", "zkxw")
try:
engine = WorkflowEngine(db)
engine.reject_task(task_id, comment=body.commentText, tenant_no=tenant_no)
return success_response(None, msg="审批驳回")
except ValueError as e:
return error_response(str(e))
注意:workflow.py 路由中 start_instance 端点直接引用了 wf_crud.WfProcessDef 和 wf_crud.WfTask,这是因为 CRUD 模块没有导出模型。更干净的做法是在路由文件顶部直接 import 模型:
from app.models.models import WfProcessDef, WfTask
并将 wf_crud.WfProcessDef 改为 WfProcessDef,wf_crud.WfTask 改为 WfTask。
- Step 2: 注册路由到 __init__.py
在 backend/app/api/__init__.py 追加:
from app.api.workflow import router as workflow_router
并在 __all__ 中添加 "workflow_router"。
- Step 3: 注册路由到 main.py
在 backend/app/main.py 中找到 app.include_router(admin_router) 等注册行,追加:
from app.api.workflow import router as workflow_router
app.include_router(workflow_router)
或者如果 main.py 从 __init__.py 导入,则追加对应 include_router 调用。
- Step 4: 验证路由加载
Run: cd backend && python -c "from app.api.workflow import router; print(f'Routes: {len(router.routes)}')"
Expected: Routes: 13
Phase 4: Frontend Foundation
Task 7: TypeScript Types + API Module
Files:
-
Modify:
web-admin/src/types/api.d.ts(追加 Workflow 命名空间) -
Create:
web-admin/src/api/workflow.ts -
Step 1: 在 api.d.ts 末尾追加 Workflow 类型
在 declare namespace Api { 的最后一个 namespace 之后(即文件最后的 } 之前)追加:
namespace Workflow {
interface ProcessDefItem {
id: string
processCode: string
processName: string
status: string
latestVersion: number | null
tenantNo: string
orderNum: number
remark: string
createTime: string
updateTime: string
}
interface VersionItem {
id: string
versionNo: number
isLatest: string
remark: string
createTime: string
updateTime: string
}
interface DesignerData {
graphJson: string | null
versionNo: number
nodes: NodeData[]
transitions: TransitionData[]
}
interface NodeData {
nodeKey: string
nodeType: string
nodeName: string | null
configJson: string | null
}
interface TransitionData {
fromNodeKey: string
toNodeKey: string
conditionExpr: string | null
}
interface TaskItem {
id: string
instanceId: string
nodeKey: string | null
assigneeUserId: string | null
status: string | null
commentText: string | null
createTime: string
updateTime: string
processName: string | null
businessType: string | null
businessId: string | null
starterUserId: string | null
instanceStatus: string | null
instanceCreateTime: string | null
}
interface InstanceDetail {
id: string
processVersionId: string
businessType: string | null
businessId: string | null
starterUserId: string | null
status: string | null
variables: string
createTime: string
tasks: TaskDetail[]
}
interface TaskDetail {
id: string
nodeKey: string | null
assigneeUserId: string | null
status: string | null
commentText: string | null
createTime: string
updateTime: string
}
}
- Step 2: 创建 workflow API 模块
Create: web-admin/src/api/workflow.ts
import request from '@/utils/http'
// ---------------------------------------------------------------------------
// 流程定义
// ---------------------------------------------------------------------------
export function fetchProcessDefList(params?: {
page?: number
pageSize?: number
keyword?: string
status?: string
}) {
return request.get<{
list: Api.Workflow.ProcessDefItem[]
total: number
page: number
pageSize: number
totalPages: number
}>({
url: '/web/admin/workflow/defs',
params,
})
}
export function fetchCreateProcessDef(data: {
processCode: string
processName: string
status?: string
orderNum?: number
remark?: string
}) {
return request.post<Api.Workflow.ProcessDefItem>({
url: '/web/admin/workflow/defs',
data,
showSuccessMessage: true,
})
}
export function fetchUpdateProcessDef(
id: string,
data: Partial<{
processCode: string
processName: string
status: string
orderNum: number
remark: string
}>,
) {
return request.put<Api.Workflow.ProcessDefItem>({
url: `/web/admin/workflow/defs/${id}`,
data,
showSuccessMessage: true,
})
}
export function fetchDeleteProcessDef(id: string) {
return request.delete({
url: `/web/admin/workflow/defs/${id}`,
showSuccessMessage: true,
})
}
// ---------------------------------------------------------------------------
// 发布版本
// ---------------------------------------------------------------------------
export function fetchPublishVersion(
defId: string,
data: {
graphJson: string
nodes: Array<{
nodeKey: string
nodeType: string
nodeName?: string | null
configJson?: string | null
}>
transitions: Array<{
fromNodeKey: string
toNodeKey: string
conditionExpr?: string | null
}>
remark?: string
},
) {
return request.post({
url: `/web/admin/workflow/defs/${defId}/publish`,
data,
showSuccessMessage: true,
})
}
export function fetchVersionList(defId: string) {
return request.get<Api.Workflow.VersionItem[]>({
url: `/web/admin/workflow/defs/${defId}/versions`,
})
}
// ---------------------------------------------------------------------------
// 设计器数据
// ---------------------------------------------------------------------------
export function fetchDesignerData(defId: string) {
return request.get<Api.Workflow.DesignerData>({
url: `/web/admin/workflow/defs/${defId}/designer`,
})
}
// ---------------------------------------------------------------------------
// 流程实例
// ---------------------------------------------------------------------------
export function fetchStartInstance(data: {
processCode: string
businessType?: string
businessId?: string
variables?: Record<string, any>
callbackUrl?: string
}) {
return request.post<{ instanceId: string }>({
url: '/web/admin/workflow/instances',
data,
})
}
export function fetchInstanceDetail(instanceId: string) {
return request.get<Api.Workflow.InstanceDetail>({
url: `/web/admin/workflow/instances/${instanceId}`,
})
}
// ---------------------------------------------------------------------------
// 任务办理
// ---------------------------------------------------------------------------
export function fetchPendingTasks(params?: {
page?: number
pageSize?: number
keyword?: string
}) {
return request.get<{
list: Api.Workflow.TaskItem[]
total: number
page: number
pageSize: number
totalPages: number
}>({
url: '/web/admin/workflow/tasks/pending',
params,
})
}
export function fetchCompletedTasks(params?: {
page?: number
pageSize?: number
keyword?: string
}) {
return request.get<{
list: Api.Workflow.TaskItem[]
total: number
page: number
pageSize: number
totalPages: number
}>({
url: '/web/admin/workflow/tasks/completed',
params,
})
}
export function fetchApproveTask(taskId: string, data?: { commentText?: string }) {
return request.post({
url: `/web/admin/workflow/tasks/${taskId}/approve`,
data,
showSuccessMessage: true,
})
}
export function fetchRejectTask(taskId: string, data?: { commentText?: string }) {
return request.post({
url: `/web/admin/workflow/tasks/${taskId}/reject`,
data,
showSuccessMessage: true,
})
}
- Step 3: 验证前端编译
Run: cd web-admin && npx vue-tsc --noEmit 2>&1 | head -20
Expected: 无新增类型错误(可能有预存错误)
Task 8: Router Entries
Files:
-
Modify:
web-admin/src/router/modules/sqxp.ts -
菜单种子:见
docs/sql/zkxw.sql内数据或docs/sql/增量脚本 -
Step 1: 在 sqxp.ts 中新增3个路由
在 sqxpAdminRoutes 数组中,找到系统管理相关的 children,追加3个路由条目:
{
id: '10000000000000000000000000000wf01',
name: 'WorkflowProcess',
path: 'workflow/process',
component: '/system/workflow/process/index',
meta: { title: '流程配置', keepAlive: true, icon: 'workflow' },
},
{
id: '10000000000000000000000000000wf02',
name: 'WorkflowPending',
path: 'workflow/pending',
component: '/system/workflow/task-pending/index',
meta: { title: '待办中心', keepAlive: true },
},
{
id: '10000000000000000000000000000wf03',
name: 'WorkflowCompleted',
path: 'workflow/completed',
component: '/system/workflow/task-completed/index',
meta: { title: '已办中心', keepAlive: true },
},
- Step 2: 创建菜单初始化SQL
在 docs/sql/zkxw.sql 或增量脚本中维护菜单数据(示例):
-- 工作流管理菜单数据
-- 将流程配置、待办中心、已办中心添加到系统管理菜单下
-- 流程配置
INSERT INTO sys_menu (id, menu_name, parent_id, order_num, path, component, menu_type, visible, status, perms, icon, menu_code, tenant_no, create_time, update_time, del_flag)
VALUES ('10000000000000000000000000000wf01', '流程配置', '0', 15, 'workflow/process', 'system/workflow/process/index', 'C', '0', '1', '', 'workflow', 'workflow:process', 'zkxw', NOW(), NOW(), 0);
-- 待办中心
INSERT INTO sys_menu (id, menu_name, parent_id, order_num, path, component, menu_type, visible, status, perms, icon, menu_code, tenant_no, create_time, update_time, del_flag)
VALUES ('10000000000000000000000000000wf02', '待办中心', '0', 16, 'workflow/pending', 'system/workflow/task-pending/index', 'C', '0', '1', '', 'bell', 'workflow:pending', 'zkxw', NOW(), NOW(), 0);
-- 已办中心
INSERT INTO sys_menu (id, menu_name, parent_id, order_num, path, component, menu_type, visible, status, perms, icon, menu_code, tenant_no, create_time, update_time, del_flag)
VALUES ('10000000000000000000000000000wf03', '已办中心', '0', 17, 'workflow/completed', 'system/workflow/task-completed/index', 'C', '0', '1', '', 'circle-check', 'workflow:completed', 'zkxw', NOW(), NOW(), 0);
Phase 5: Flow Designer
Task 9: AntV X6 Flow Designer
Files:
- Create:
web-admin/src/views/system/workflow/designer/FlowDesigner.vue
前置条件: 安装 AntV X6 依赖
- Step 1: 安装 AntV X6
Run: cd web-admin && pnpm add @antv/x6 @antv/x6-plugin-selection @antv/x6-plugin-dnd @antv/x6-plugin-minimap @antv/x6-plugin-history
- Step 2: 创建 FlowDesigner.vue
这是一个大型组件,实现三栏布局(节点面板 / 画布 / 属性面板)。
关键实现要点:
- 使用
@antv/x6的Graph类初始化画布 - 注册4种自定义节点(Start/End/Approval/Condition)使用
Graph.registerNode - 左侧面板提供拖拽源,使用
@antv/x6-plugin-dnd实现拖入画布 - 选中节点时右侧属性面板展示对应配置表单
- 保存时序列化为
{graphJson, nodes, transitions}格式 - 发布调用
fetchPublishVersionAPI
组件 props:
const props = defineProps<{
visible: boolean
processDefId: string
processName: string
}>()
const emit = defineEmits<{
(e: 'update:visible', val: boolean): void
(e: 'saved'): void
}>()
组件核心结构:
<template>
<el-dialog
:model-value="visible"
:title="`流程设计 - ${processName}`"
width="100%"
fullscreen
destroy-on-close
@update:model-value="$emit('update:visible', $event)"
class="flow-designer-dialog"
>
<div class="flow-designer">
<!-- 左侧节点面板 -->
<div class="designer-sidebar">
<div class="sidebar-title">节点组件</div>
<div
v-for="node in nodeTypes"
:key="node.type"
class="sidebar-node"
:style="{ borderColor: node.color }"
draggable="true"
@dragstart="onDragStart($event, node)"
>
<div class="sidebar-node-icon" :style="{ backgroundColor: node.color }">
{{ node.icon }}
</div>
<span>{{ node.label }}</span>
</div>
</div>
<!-- 中间画布 -->
<div class="designer-canvas" ref="canvasRef"></div>
<!-- 右侧属性面板 -->
<div class="designer-props">
<div v-if="selectedNode" class="props-content">
<h3>节点属性</h3>
<el-form label-position="top">
<el-form-item label="节点名称">
<el-input v-model="selectedNode.data.nodeName" @change="updateNodeData" />
</el-form-item>
<!-- 审批人节点配置 -->
<template v-if="selectedNode.data.nodeType === 'approval'">
<el-form-item label="审批人策略">
<el-select v-model="approvalConfig.strategy" @change="updateNodeData">
<el-option label="指定用户" value="users" />
<el-option label="指定角色" value="roles" />
<el-option label="SQL查询" value="sql" />
</el-select>
</el-form-item>
<el-form-item v-if="approvalConfig.strategy === 'users'" label="选择用户">
<el-select v-model="approvalConfig.user_ids" multiple @change="updateNodeData">
<!-- 动态加载用户列表 -->
</el-select>
</el-form-item>
<el-form-item v-if="approvalConfig.strategy === 'roles'" label="选择角色">
<el-select v-model="approvalConfig.role_ids" multiple @change="updateNodeData">
<!-- 动态加载角色列表 -->
</el-select>
</el-form-item>
<el-form-item v-if="approvalConfig.strategy === 'sql'" label="SQL语句">
<el-input
v-model="approvalConfig.sql"
type="textarea"
:rows="4"
placeholder="SELECT user_id FROM ... WHERE dept_id = :dept_id"
@change="updateNodeData"
/>
</el-form-item>
<el-form-item label="通过策略">
<el-radio-group v-model="approvalConfig.pass_type" @change="updateNodeData">
<el-radio value="any">任一通过即可</el-radio>
<el-radio value="all">全部通过才行</el-radio>
</el-radio-group>
</el-form-item>
</template>
<!-- 条件分支配置(在出边上配置,此处提示) -->
<template v-if="selectedNode.data.nodeType === 'condition'">
<p class="props-hint">条件规则在连线上配置。点击连线设置条件表达式。</p>
</template>
</el-form>
</div>
<div v-else class="props-empty">
<p>选中节点以编辑属性</p>
</div>
</div>
</div>
<template #footer>
<el-button @click="$emit('update:visible', false)">取消</el-button>
<el-button class="btn-ghost" @click="handleSaveDraft">保存草稿</el-button>
<el-button class="btn-main" type="primary" @click="handlePublish">发布新版本</el-button>
</template>
</el-dialog>
</template>
节点类型定义:
const nodeTypes = [
{ type: 'start', label: '开始节点', icon: '▶', color: '#10b981' },
{ type: 'end', label: '结束节点', icon: '⏹', color: '#ef4444' },
{ type: 'approval', label: '审批人节点', icon: '👤', color: '#3b82f6' },
{ type: 'condition', label: '条件分支', icon: '◆', color: '#f59e0b' },
]
核心方法:
// 初始化画布
function initGraph() {
const graph = new Graph({
container: canvasRef.value,
grid: true,
mousewheel: { enabled: true, modifiers: ['ctrl', 'meta'] },
connecting: {
router: 'manhattan',
connector: { name: 'rounded', args: { radius: 8 } },
anchor: 'center',
connectionPoint: 'anchor',
allowBlank: false,
allowLoop: false,
allowMulti: true,
snap: { radius: 30 },
},
highlighting: {
magnetAdsorbed: { name: 'stroke', args: { attrs: { fill: '#10b981', stroke: '#10b981' } } },
},
history: true,
})
// 选中节点事件
graph.on('node:click', ({ node }) => {
selectedNode.value = node
})
graph.on('blank:click', () => {
selectedNode.value = null
})
graphRef.value = graph
}
// 序列化
function serializeGraph() {
const graph = graphRef.value
if (!graph) return null
const cells = graph.getCells()
const nodes = cells
.filter((c) => c.isNode())
.map((n) => ({
nodeKey: n.id,
nodeType: n.getData()?.nodeType || 'start',
nodeName: n.getData()?.nodeName || '',
configJson: JSON.stringify(n.getData()?.config || {}),
}))
const edges = cells
.filter((c) => c.isEdge())
.map((e) => ({
fromNodeKey: (e.getSourceCellId() || ''),
toNodeKey: (e.getTargetCellId() || ''),
conditionExpr: e.getData()?.conditionExpr || null,
}))
return {
graphJson: JSON.stringify(graph.toJSON()),
nodes,
transitions: edges,
}
}
- Step 3: 验证设计器组件编译
Run: cd web-admin && npx vue-tsc --noEmit 2>&1 | grep -i "FlowDesigner" || echo "No FlowDesigner errors"
Expected: 无 FlowDesigner 相关错误
Phase 6: Frontend Pages
Task 10: Process Management Page
Files:
-
Create:
web-admin/src/views/system/workflow/process/index.vue -
Step 1: 创建流程配置管理页面
参考 unit/index.vue 标准列表模板。功能:列表 + 新增/编辑弹窗 + 流程设计器弹窗。
列表字段:流程编码、流程名称、状态(启用/停用 switch)、最新版本、创建时间、操作(设计/编辑/删除)。
操作列宽度:4按钮(设计/编辑/停用启用/删除)= 400px。
关键交互:
- "流程设计"按钮 → 打开 FlowDesigner 全屏弹窗
- "停用/启用"切换 → 调用
fetchUpdateProcessDef修改 status - 新增/编辑弹窗使用
.form-grid-2+label-position="top"
Task 11: Approval Dialog + Status Tag Components
Files:
-
Create:
web-admin/src/views/system/workflow/components/ApprovalDialog.vue -
Create:
web-admin/src/views/system/workflow/components/ProcessStatusTag.vue -
Step 1: 创建审批弹窗组件
<template>
<el-dialog
v-model="visible"
title="审批操作"
width="480px"
destroy-on-close
class="dialog-form-layout"
align-center
>
<el-form label-position="top">
<el-form-item label="审批结果">
<el-radio-group v-model="form.approved">
<el-radio :value="true">通过</el-radio>
<el-radio :value="false">驳回</el-radio>
</el-radio-group>
</el-form-item>
<el-form-item label="审批意见">
<el-input
v-model="form.commentText"
type="textarea"
:rows="3"
placeholder="请输入审批意见"
/>
</el-form-item>
</el-form>
<template #footer>
<el-button @click="visible = false">取消</el-button>
<el-button type="primary" @click="handleSubmit">确认提交</el-button>
</template>
</el-dialog>
</template>
- Step 2: 创建状态标签组件
<template>
<el-tag :type="tagType" size="small">{{ label }}</el-tag>
</template>
<script setup lang="ts">
const props = defineProps<{ status: string }>()
const statusMap: Record<string, { label: string; type: string }> = {
running: { label: '进行中', type: 'warning' },
completed: { label: '已完成', type: 'success' },
rejected: { label: '已驳回', type: 'danger' },
cancelled: { label: '已取消', type: 'info' },
pending: { label: '待处理', type: 'warning' },
approved: { label: '已通过', type: 'success' },
withdrawn: { label: '已撤回', type: 'info' },
}
const tagType = computed(() => statusMap[props.status]?.type || 'info')
const label = computed(() => statusMap[props.status]?.label || props.status)
</script>
Task 12: Pending Tasks Page
Files:
-
Create:
web-admin/src/views/system/workflow/task-pending/index.vue -
Step 1: 创建待办中心页面
参考 order/index.vue 复杂列表模板。
筛选:流程名称关键词、发起时间范围(el-date-picker type="daterange")。
列表字段:流程名称、业务类型、业务单号、发起人、到达时间、操作。
操作列:审批(打开ApprovalDialog)、查看详情(展开行或弹窗显示实例详情含任务列表)。
操作列宽度:2按钮 = 260px。
Task 13: Completed Tasks Page
Files:
-
Create:
web-admin/src/views/system/workflow/task-completed/index.vue -
Step 1: 创建已办中心页面
参考标准列表模板。
筛选:流程名称关键词、审批结果(通过/驳回 select)。
列表字段:流程名称、业务类型、业务单号、发起人、审批结果(ProcessStatusTag)、审批意见、审批时间。
操作列:查看详情。1按钮 = 260px。
Phase 7: Integration
Task 14: App.vue Title Whitelist + Final Verification
Files:
-
Modify:
web-admin/src/App.vue -
Step 1: 在 App.vue panelTitleBarTextRules 中追加工作流标题
在 panelTitleBarTextRules 数组中追加:
'流程配置',
'待办列表',
'已办列表',
- Step 2: 运行后端验证
Run: cd backend && python -c "from app.main import app; routes = [r.path for r in app.routes if hasattr(r, 'path')]; wf = [r for r in routes if 'workflow' in r]; print(f'Workflow routes: {len(wf)}'); [print(f' {r}') for r in wf]"
Expected: 13 workflow routes listed
- Step 3: 运行前端编译检查
Run: cd web-admin && pnpm build 2>&1 | tail -5
Expected: 构建成功(或仅有预存错误)
- Step 4: Commit
git add backend/ web-admin/ docs/
git commit -m "feat: add workflow module (2.1.11) - process config, designer, task centers"