向导链接工作流管理器
workflow_manager的Python项目详细描述
#Workflow Manager基于任务的Workflow Manager的Python实现。
此软件包允许轻松包装任何依赖于代码库中其他功能的功能。
此包允许在向导中实例化所有任务规则,然后使用一个简单的管理器包装器在一次调用中执行工作流。
此包还提供查看工作流历史的功能,以便进行调试。
两个
任务3
任务4
任务5
任务6
````
以及以下业务规则:
````
如果任务1成功,任务2和任务3将按顺序执行
如果任务1失败,任务5将执行
如果任务2成功,任务4将执行
如果任务2失败,任务5将执行
如果任务3失败,任务6将执行
如果任务4失败,任务6将执行
如果任务5成功,任务6将执行
````
将以正确的顺序执行正确的依赖项。
模块还将短路故障情况下的任何调用,但将执行完全清理工作流所需的所有故障依赖项。
有关具体示例,查看“tests/test_workflow.py”。
用自己的逻辑覆盖“execute”方法:
``python
from workflow\u manager.task import task
class customtask(task):
def\uu init\uu(self):
super()。
def execute(self,**kwargs):
(task.success\u state(),'result','in','a','list')
否则:
返回(task.failure\u state(),'this failed because of reasons')
class anothertask(task):
super()。
def execute(self,**kwargs):
如果成功:
返回(task.success_state(),'result','in','a','list')
否则:
返回(task.failure_state(),'this failed because of reasons')
````
,添加您的业务规则。
``python
customtask=customtask()
anothertask=anothertask()
customtask.onu success(anothertask,其他任务)
customtask.on_failure(cleanuptask)
anothertask.on_success(keepitgoingtask)
anothertask.on_failure(cleanuptask)
``````
json
customtask.to_dict();返回工作流的字典
```
并调用“run”函数:
``python
from workflow\u manager.manager import manager
manager.register初始任务(customtask)
````
如果要查看工作流结束后发生的事情,可以调用“show\u executed\u flow”方法,该方法将返回任务列表和参数。
`manager.show_executed_flow()`
此软件包允许轻松包装任何依赖于代码库中其他功能的功能。
此包允许在向导中实例化所有任务规则,然后使用一个简单的管理器包装器在一次调用中执行工作流。
此包还提供查看工作流历史的功能,以便进行调试。
两个
任务3
任务4
任务5
任务6
````
以及以下业务规则:
````
如果任务1成功,任务2和任务3将按顺序执行
如果任务1失败,任务5将执行
如果任务2成功,任务4将执行
如果任务2失败,任务5将执行
如果任务3失败,任务6将执行
如果任务4失败,任务6将执行
如果任务5成功,任务6将执行
````
将以正确的顺序执行正确的依赖项。
模块还将短路故障情况下的任何调用,但将执行完全清理工作流所需的所有故障依赖项。
有关具体示例,查看“tests/test_workflow.py”。
用自己的逻辑覆盖“execute”方法:
``python
from workflow\u manager.task import task
class customtask(task):
def\uu init\uu(self):
super()。
def execute(self,**kwargs):
(task.success\u state(),'result','in','a','list')
否则:
返回(task.failure\u state(),'this failed because of reasons')
class anothertask(task):
def execute(self,**kwargs):
如果成功:
返回(task.success_state(),'result','in','a','list')
否则:
返回(task.failure_state(),'this failed because of reasons')
````
,添加您的业务规则。
``python
customtask=customtask()
anothertask=anothertask()
customtask.onu success(anothertask,其他任务)
customtask.on_failure(cleanuptask)
anothertask.on_success(keepitgoingtask)
anothertask.on_failure(cleanuptask)
``````
json
customtask.to_dict();返回工作流的字典
```
并调用“run”函数:
``python
from workflow\u manager.manager import manager
如果要查看工作流结束后发生的事情,可以调用“show\u executed\u flow”方法,该方法将返回任务列表和参数。
`manager.show_executed_flow()`