SpiffWorkflow 自定义 Task Spec 实战：从类实现、转换器到解析器覆盖的完整流程

后端工作流自动化流程编排低代码【免费下载链接】SpiffWorkflowA powerful workflow engine implemented in pure Python项目地址https://gitcode.com/gh_mirrors/sp/SpiffWorkflow点击查看免费下载本文基于 SpiffWorkflow 官方文档 Implementing a Custom Task Spec 展开完整讲解如何用自定义 Task Spec 类替换 BPMN 内置任务行为以接管 Timer Start Event 为例并结合当前仓库源码逐一剖析类继承关系、序列化 Converter、SPIFF_CONFIG 注册与OVERRIDE_PARSER_CLASSES解析器覆盖四处关键机制。读完本文你可以独立实现一个可序列化、可解析、可运行的自定义任务节点。SpiffWorkflow 是一个纯 Python 实现的 BPMN 工作流引擎。当流程中存在长等待场景时例如以 Duration Timer 开始的事件要等一天、数天甚至数周才触发你可能不希望引擎自己管理这段等待时间而是把控制权交给外部系统。官方文档给出的方案就是Custom Task Spec自定义一个任务规格类、配套的序列化转换器并覆盖解析器中该 BPMN 元素对应的类映射四步即可让引擎按你的逻辑处理特定节点。问题背景Timer Start Event 的长等待假设你有一个以 Timer Start Event 开头的流程流程加载后Timer 开始事件会一直等待计时到期这可能长达数天。官方文档指出常规做法是确认它处于等待状态后把 workflow 序列化、等到时间到了再恢复。但如果你根本不想让 SpiffWorkflow 管理这个等待就需要自定义 Task Spec。官方文档中的示例代码位于示例项目spiff-example-cli的misc/custom_start_event.py中配套的 BPMN 模型bpmn/tutorial/timer_start.bpmn的 process ID 为timer_startStart Event 上挂了一个一小时的 Duration Timer一天。在示例应用中使用默认配置运行时会看到一个处于WAITING状态的 Start Event而使用文档中创建的自定义 parser 和 serializer 后后续 User Task 会立刻进入可完成状态。在示例项目中运行该模型的命令如下注意-e参数指定了自定义模块./runner.py -e spiff_example.misc.custom_start_event add -p timer_start -b bpmn/tutorial/timer_start.bpmn ./runner.py -e spiff_example.misc.custom_start_event说明runner.py、FileSerializer等属于示例项目spiff-example-cli的配套代码不在本仓库中本仓库提供的是它们所依赖的全部核心类与扩展点。其中FileSerializer是对本仓库 BpmnWorkflowSerializer 的文件落盘封装详见 序列化文档。整个改造分为四步定义自定义 Spec 类 → 实现 Converter → 注册到序列化配置 → 覆盖解析器映射。下面逐步展开。第一步定义自定义 Task Spec 类核心思路在构造时检查传入的 event definition如果它是TimerEventDefinition就把它替换成NoneEventDefinition同时把原 timer 定义保存到一个新属性timer_event上供外部使用。from SpiffWorkflow.bpmn.specs.event_definitions import NoneEventDefinition from SpiffWorkflow.bpmn.specs.event_definitions.timer import TimerEventDefinition from SpiffWorkflow.bpmn.specs.mixins import StartEventMixin from SpiffWorkflow.spiff.specs import SpiffBpmnTask class CustomStartEvent(StartEventMixin, SpiffBpmnTask): def __init__(self, wf_spec, bpmn_id, event_definition, **kwargs): if isinstance(event_definition, TimerEventDefinition): super().__init__(wf_spec, bpmn_id, NoneEventDefinition(), **kwargs) self.timer_event event_definition else: super().__init__(wf_spec, bpmn_id, event_definition, **kwargs) self.timer_event None这段代码中有几个设计要点都可以从仓库源码中找到依据为什么判断基类TimerEventDefinition。Timer 事件有三类子类定义在 SpiffWorkflow/bpmn/specs/event_definitions/timer.pyTimeDateEventDefinitionL144具体时间触发、DurationTimerEventDefinitionL160持续时间触发has_fired基于datetime.now(timezone.utc)比较、CycleTimerEventDefinitionL178ISO 8601 循环间隔触发。统一判断基类TimerEventDefinitionL13即可一次性覆盖 TimeDate、Duration、Cycle 三种情况这正是官方文档 note 中强调的做法。为什么替换成NoneEventDefinition后流程会立即推进。等待事件的通用行为由 CatchingEvent 的_update_hook实现def _update_hook(self, my_task): super()._update_hook(my_task) # None events dont propagate, so as soon as were ready, we fire our event if isinstance(self.event_definition, NoneEventDefinition): my_task._set_internal_data(event_firedTrue) if self.event_definition.has_fired(my_task): return True elif my_task.state ! TaskState.WAITING: my_task._set_state(TaskState.WAITING) my_task.workflow.top_workflow.event_manager.add_task(my_task) self.event_definition.update_task(my_task)可以看到当 event definition 是NoneEventDefinition时任务直接就标记event_firedTrue并通过has_fired返回 True任务不会进入WAITING、也不会注册到event_manager。而 timer 定义如DurationTimerEventDefinition.has_fired则会在到期前持续返回 False任务保持WAITING——这就是「替换后 User Task 立刻可完成」的底层原因。双继承结构的由来。StartEventMixin即 SpiffWorkflow/bpmn/specs/mixins/events/start_event.py 中的StartEvent通过 mixins/init.py 以别名导出它继承CatchingEvent并覆写了catch如果 start event 已经完成或被取消后又收到事件会调用workflow.reset_from_task_id(my_task.id)重置流程。SpiffBpmnTask定义在 SpiffWorkflow/spiff/specs/spiff_task.py它在BpmnSpecMixin基础上增加了prescript/postscript脚本钩子。class SpiffBpmnTask(BpmnSpecMixin): def __init__(self, wf_spec, name, prescriptNone, postscriptNone, **kwargs): super().__init__(wf_spec, name, **kwargs) self.prescript prescript self.postscript postscript官方文档特别说明SpiffWorkflow 已经把各 BPMN 任务的基础行为从BpmnSpecMixin中拆分成独立的 mixin就是为了让你可以这样组合继承而不陷入 MRO方法解析顺序问题。作为对照内置默认实现 SpiffWorkflow/bpmn/specs/defaults.py 中的StartEvent(StartEventMixin, BpmnSpecMixin)用的正是同样的双继承模式。经验法则实现自定义 task spec 时通常需要从「BPMN 行为 mixin」和「spiff 基础类」两个类别各继承一个基类。官方文档同时留有一个设计上的 note更理想的构造函数应该同时接收要使用的 event definition 和原始的 timer event definition但由于现有 parser 不够直观、不易扩展示例采用了上面这种「先替换、再挂载」的写法以便阅读。第二步实现序列化 Converter每一个自定义 task spec 都必须配套一个 converter否则序列化器不知道如何持久化它的自定义属性。文档中的实现如下from SpiffWorkflow.bpmn.serializer import BpmnWorkflowSerializer from SpiffWorkflow.spiff.serializer.task_spec import SpiffBpmnTaskConverter from SpiffWorkflow.spiff.serializer import DEFAULT_CONFIG class CustomStartEventConverter(SpiffBpmnTaskConverter): def to_dict(self, spec): dct super().to_dict(spec) dct[event_definition] self.registry.convert(spec.event_definition) dct[timer_event] self.registry.convert(spec.timer_event) return dct def from_dict(self, dct): spec super().from_dict(dct) spec.event_definition self.registry.restore(dct[event_definition]) spec.timer_event self.registry.restore(dct[timer_event]) return spec转换器约定to_dict接收一个 task spec、返回 JSON 可序列化的字典from_dict接收字典、还原出对应类型的 task spec。实现方式几乎都是「先调父类完成大部分工作再补上自己新增的属性」。从源码看这条继承链SpiffBpmnTaskConverter 的to_dict在默认属性之外追加了prescript/postscript两个字段from_dict则委托给基类的task_spec_from_dictclass SpiffBpmnTaskConverter(TaskSpecConverter): def to_dict(self, spec): dct self.get_default_attributes(spec) dct[prescript] spec.prescript dct[postscript] spec.postscript return dct def from_dict(self, dct): return self.task_spec_from_dict(dct)基类 TaskSpecConverter 位于bpmn.serializer.helpers包文档所说的 similar base converter。它的get_default_attributesL125-L151负责抽取标准 BPMN 属性name、description、manual、lookahead、inputs、outputs、bpmn_id、bpmn_name、lane、documentation以及通过self.registry.convert序列化data_input_associations、data_output_associations、io_specificationtask_spec_from_dictL195-L223反向重建对象包括用self.target_class(wf_spec, name, **dct)实例化并处理bpmn_id ! name的多实例边界情况。self.registry.convert(spec.timer_event)之所以能直接工作是因为三种 timer event definition 在默认配置里早已注册SpiffWorkflow/bpmn/serializer/config.py 中TimeDateEventDefinition、DurationTimerEventDefinition、CycleTimerEventDefinition都映射到TimerConditionalEventDefinitionConverter。你不需要为它们重复注册只需把自定义 spec 上新出现的属性这里是timer_event显式交给 registry 即可。第三步把转换器注册进序列化配置序列化器初始化时接收一个「对象类 → 转换器类」的映射表因此注册自定义 spec 只需加一条映射SPIFF_CONFIG[CustomStartEvent] CustomStartEventConverter registry FileSerializer.configure(SPIFF_CONFIG) serializer FileSerializer(dirname, registryregistry)这里的SPIFF_CONFIG来自 SpiffWorkflow/spiff/serializer/config.py。它的构建方式本身就是一个值得参考的注册表模式先对bpmn包的DEFAULT_CONFIG做deepcopy然后pop掉被 spiff 版本替换掉的默认任务类再把UserTask、ScriptTask、ServiceTask、各 event definition 等逐一映射到对应 converter。你在自己的模块中只需在这个副本上追加自己的条目。configure的底层逻辑在 BpmnWorkflowSerializer.configurestaticmethod def configure(configNone, registryNone): config config or DEFAULT_CONFIG if registry is None: registry DefaultRegistry() for target_class, converter_class in config.items(): converter_class(target_class, registry) return registry它遍历配置表用(target_class, registry)实例化每个 converter 并挂入DefaultRegistryregistry 之后通过convert/restore统一完成对象与字典的互转。类 docstring 也明确指出如果你替换了 Spiff 使用的默认类而没有提供转换机制「Spiff 将不知道如何序列化它们」——这正是必须执行第三步的原因。另外两个与版本兼容相关的细节serialize_json会把VERSION_KEY serializer_version写入输出当前默认版本为1.4workflow.py L29反序列化时会经migrate按版本执行迁移函数。自定义 converter 时保持to_dict/from_dict的字段对称性是将来跨版本迁移不丢数据的前提。第四步覆盖解析器的类映射光有自定义 spec 还不够——解析 BPMN XML 时引擎默认仍会创建内置的StartEvent对象。必须告诉解析器改用你的类from SpiffWorkflow.spiff.parser import SpiffBpmnParser from SpiffWorkflow.spiff.parser.event_parsers import StartEventParser from SpiffWorkflow.bpmn.parser.util import full_tag parser SpiffBpmnParser() parser.OVERRIDE_PARSER_CLASSES[full_tag(startEvent)] (StartEventParser, CustomStartEvent)这背后的机制在 BpmnParser 中PARSER_CLASSESL125-L146是内置映射表例如full_tag(startEvent): (StartEventParser, StartEvent)OVERRIDE_PARSER_CLASSESL148默认为空字典作为扩展点存在取解析类时优先查 override 表def _get_parser_class(self, tag): if tag in self.OVERRIDE_PARSER_CLASSES: return self.OVERRIDE_PARSER_CLASSES[tag] elif tag in self.PARSER_CLASSES: return self.PARSER_CLASSES[tag] return None, None值是一个二元组(TaskParser 类, TaskSpec 类)。full_tag是 SpiffWorkflow/bpmn/parser/util.py 提供的小工具把短标签补全为带命名空间http://www.omg.org/spec/BPMN/20100524/MODEL的完整 tag例如{http://www.omg.org/spec/BPMN/20100524/MODEL}startEvent。SpiffBpmnParser 本身就大量使用了这个机制——它的OVERRIDE_PARSER_CLASSES把task、userTask、scriptTask、serviceTask、startEvent等元素全部映射到 Spiff 扩展版解析器与 Spiff 任务类其中full_tag(startEvent): (SpiffStartEventParser, StartEvent)即 L91。也就是说「在实例上更新映射值而非子类化 parser」是本项目解析器既定的扩展风格官方文档也坦言这一点「parser 用类属性定义如何解析某个元素以及用什么类创建 task spec所以这里不是传参而是创建 parser 后更新它的值。这有点不直观但就是这么运作的。」值得庆幸的是本例可以直接复用现有的 StartEventParserspiff 侧的 SpiffStartEventParser 也只是在事件定义解析之外包装了它只需要替换二元组中的 spec 类省去了自己写 XML 解析的工作。测试代码 tests/SpiffWorkflow/bpmn/BpmnLoaderForTests.py 同样通过重写OVERRIDE_PARSER_CLASSES注入测试用解析类可见这是仓库内部认可的标准扩展方式。端到端串联数据流与状态流回顾把四步串起来一次完整运行涉及两条链路解析期SpiffBpmnParser读取 BPMN XML →_get_parser_class(full_tag(startEvent))命中 override → 用StartEventParser解析事件定义、用CustomStartEvent创建 spec → 构造函数检测到TimerEventDefinition后换成NoneEventDefinition并保留timer_event属性。运行期CatchingEvent._update_hook因NoneEventDefinition立即置event_firedTrue任务不进入WAITING流程直接推进到下一个 User Task。持久化期BpmnWorkflowSerializer按 registry 找到CustomStartEventConverterto_dict输出标准属性 event_definitiontimer_event两份字典from_dict用registry.restore把两者还原为具体的 event definition 对象恢复出一个与原对象行为一致的CustomStartEvent。如果三步中漏掉任何一步都会出现不同症状漏掉 Converter 注册时registry 不认识该 spec序列化阶段无法生成字典漏掉 parser 覆盖时运行时对象仍是内置StartEventconverter 永远不会被触发而 Converter 中漏掉timer_event字段则反序列化后的对象与解析时的属性不一致。实践要点小结继承模式自定义 spec 一般同时继承「bpmn 行为 mixin」如StartEventMixin与「spiff 基础类」如SpiffBpmnTask参考内置 defaults.py 的StartEvent(StartEventMixin, BpmnSpecMixin)写法避免 MRO 问题。转换器基线优先继承与你 spec 基类对应的 converter本例为 SpiffBpmnTaskConverter只补新增属性标准属性由 TaskSpecConverter.get_default_attributes 统一处理。注册在SPIFF_CONFIG或自己deepcopy(DEFAULT_CONFIG)的副本中追加spec 类 - converter 类映射再经BpmnWorkflowSerializer.configure生成 registry。解析覆盖对 parser 实例的OVERRIDE_PARSER_CLASSES写入full_tag(元素名) - (parser 类, spec 类)优先复用现有 parser。可序列化前提spec 上任何需要持久化的对象属性本例的两个 event definition都要能走registry.convert/restoretimer 三种定义已在默认配置中注册直接可用。以上流程与官方文档 docs/bpmn/custom_task_spec.rst 完全对应配套的序列化细节可继续阅读 docs/bpmn/serialization.rst解析器与自定义对象注册registry.register等机制也在该文档中有更完整的展开。赞分享后端工作流自动化流程编排低代码【免费下载链接】SpiffWorkflowA powerful workflow engine implemented in pure Python项目地址https://gitcode.com/gh_mirrors/sp/SpiffWorkflow点击查看免费下载相关推荐京豆收益最大化策略JD_tencent_scf 定时任务与脚本组合实战京豆收益最大化策略JD_tencent_scf 定时任务与脚本组合实战 想让京东京豆收益最大化只靠手动签到远远不够。JD_tencent_scf 是一套开箱CsvHelper 自定义类型转换器Custom Type Converters完整实战指南CsvHelper 自定义类型转换器Custom Type Converters完整实战指南 CsvHelper 内置的类型转换器已经覆盖了绝大多数 .NE后端数据工程如何快速上手Meltano从安装到第一个ELT管道如何快速上手Meltano从安装到第一个ELT管道 Meltano是一款声明式代码优先的数据集成引擎能帮助你轻松实现数据和机器学习产品的集成需求无需编写、创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考