模型抽象

上文对工作流进行了一些基本的介绍，有了基础篇的内容，相信我们可以很容易抽象了一些实体模型：流程模型、开始节点模型、结束节点模型、任务节点模型、决策节点模型、分支节点模型、合并节点模型、边模型。

基础模型

我们把模型进行进一步抽象，可以得到基础模型，基础模型只包含两个属性，唯一编码和显示名称。

class BaseModel (object):
    """
    基础模型
    """
    __abstract__ = True
    # 唯一编码
    name = None
    # 显示名称
    displayName = None

模型行为

如果按照面向对象来看，流程模型本身除了属性外，还应该具有方法和行为，即节点和边所能做的事，这里用Action接口来定义。但是Python中没有接口，所以这里我们使用抽象类来定义。

class Action (object):
    """
    节点行为接口
    """
    __abstract__ = True
    def execute(self, execution):
        """
        执行行为
        @param execution: 执行对象参数
        """

节点模型

流程模型由节点模型和边构成，节点的类型虽然很多，但是依然可以对其进行进一步抽象。节点都有自己通用性的行为，如前置拦截、后置拦截、日志处理等，除了通用性行为外，还可以根据子类不同的特性，实现exec抽象方法。在python中，实现抽象方法，需要使用abc模块，并且使用@abstractmethod装饰器。

from abc import ABC, abstractmethod
class NodeModel (ABC, BaseModel, Action):
    """
    节点模型
    """
    __abstract__ = True
    # 输入边集合
    inputs = []
    # 输出边集合
    outputs = []
    # 前置拦截器-字符串，多个使用英文逗号分割
    preInterceptors = None
    # 后置拦截器-字符串，多个使用英文逗号分割
    postInterceptors = None
    @abstractmethod
    def exec(self, execution):
        """
        由子类自定义执行方法
        @param execution: 执行对象参数
        """
    def execute(self, execution):
        """
        执行节点
        @param execution: 执行对象参数
        """
        self.exec(execution)

有了节点模型后，其他具体的节点模型就会显示相对简单。

开始节点模型

开始节点主要继承节点模型并实现其exec方法。

class StartModel(NodeModel):
    """
    开始节点模型
    """
    def exec(self, execution):
        """
        执行节点
        @param execution: 执行对象参数
        """
        # 执行开始节点自定义执行逻辑

结束节点模型

结束节点主要继承节点模型并实现其exec方法。

class EndModel(NodeModel):
    """
    开始节点模型
    """
    def exec(self, execution):
        """
        执行节点
        @param execution: 执行对象参数
        """
        # 执行结束节点自定义执行逻辑

任务节点模型

任务节点是流程中非常重要的节点，其拥有更多自己的属性。当然，下面只是把常用的属性罗列出来，后续也根据具体实现的业务，增加和删除其属性。

class TaskModel(NodeModel):
    """
    任务节点模型
    """
    form = None # 表单标识
    assignee = None # 参与人
    assignmentHandler = None # 参与人处理类
    taskType = None # 任务类型(主办/协办)
    performType = None # 参与类型(普通参与/会签参与)
    reminderTime = None # 提醒时间
    reminderRepeat = None # 重复提醒间隔
    expireTime = None # 期待任务完成时间变量key
    autoExecute = None # 到期是否自动执行Y/N
    callback = None # 自动执行回调类
    ext = {} # 自定义扩展属性
    def exec(self, execution):
        """
        执行节点
        @param execution: 执行对象参数
        """
        # 执行任务节点自定义执行逻辑

决策节点模型

决策节点也是一种特殊节点，其会有决策表达式，和决策处理类属性，主要用于动态计算下一个节点的走向。

class DecisionModel(NodeModel):
    """
    决策节点模型
    """
    def exec(self, execution):
        """
        执行节点
        @param execution: 执行对象参数
        """
        # 执行决策节点自定义执行逻辑

分支节点模型

当用到并行流程的时候，该节点就派上用场了，这里只是先简单定义，后续也会根据具体的业务需要，扩展新的属性。

class ForkModel(NodeModel):
    """
    分支节点模型
    """
    def exec(self, execution):
        """
        执行节点
        @param execution: 执行对象参数
        """
        # 执行分支节点自定义执行逻辑

合并节点模型

合并节点和分支节点是一起出来的，这里也是先按照最基础的定义。

class JoinModel(NodeModel):
    """
    合并节点模型
    """
    def exec(self, execution):
        """
        执行节点
        @param execution: 执行对象参数
        """
        # 执行合并节点自定义执行逻辑

边模型

两节点连线，就构成一条边，所以边的关系属性就是source和target，边也有自己的执行方法。

class TransitionModel(BaseModel,Action):
    source = None # 边源节点引用
    target = None # 边目标节点引用
    to = None # 目标节点名称
    expr = None # 边表达式
    enabled = False # 是否可执行

流程模型

有了上面的元素，那么流程模型就很容易得到如下定义：

class ProcessModel(BaseModel):
    type = None # 流程定义分类
    instanceUrl = None # 启动实例要填写的表单key
    expireTime = None # 期待完成时间变量key
    instanceNoClass = None # 实例编号生成器实现类
    # 流程定义的所有节点
    nodes = []
    # 流程定义的所有任务节点
    tasks = []

实例流程模型对象

简单模型

from packages.engine.model.end_model import EndModel
from packages.engine.model.process_model import ProcessModel
from packages.engine.model.start_model import StartModel
import logging

from packages.engine.model.task_model import TaskModel
from packages.engine.model.transition_model import TransitionModel


class TestBuildProcess:
    def test_build(self):
        processModel = ProcessModel()
        startModel = StartModel()
        startModel.name = 'start'
        startModel.displayName = '开始'
        processModel.nodes.append(startModel)
        t1 = TransitionModel()
        t1.name = 't1'
        t1.to = 'apply'
        t1.source = startModel
        startModel.outputs.append(t1)
        applyTaskModel = TaskModel()
        t1.target = applyTaskModel
        applyTaskModel.name = 'apply'
        applyTaskModel.displayName = '请假申请'
        processModel.nodes.append(applyTaskModel)
        processModel.tasks.append(applyTaskModel)
        t2 = TransitionModel()
        t2.name = 't2'
        t2.to = 'deptApprove'
        t2.source = 'applyTaskModel'
        applyTaskModel.outputs.append(t2)

        deptApproveTaskModel =TaskModel()
        t2.target = deptApproveTaskModel
        deptApproveTaskModel.name = 'deptApprove'
        deptApproveTaskModel.displayName = '部门领导审批'
        processModel.nodes.append(deptApproveTaskModel)
        processModel.tasks.append(deptApproveTaskModel)

        t3 = TransitionModel()
        t3.name = 't3'
        t3.to = 'end'
        t3.source = deptApproveTaskModel
        deptApproveTaskModel.outputs.append(t3)

        endModel = EndModel()
        t3.target = endModel
        endModel.name = 'end'
        endModel.displayName = '结束'
        processModel.nodes.append(endModel)
        for node in processModel.nodes:
            logging.info(f"{node.name},{node.displayName}")

执行单元测试

pytest -q tests/test_build_process.py

执行结果如下：

执行结果

相关源码

工作流引擎核心设计·模型抽象open in new window