Task Groups¶
Use group_name to run independent branches of tasks in grouped execution flows.
Example¶
from time import sleep
from dotflow import DotFlow, action
@action
def task_foo():
sleep(5)
print("2: Task Foo")
return "ok"
@action
def task_bar():
sleep(2)
print("1: Task Bar")
return "ok"
def main():
workflow = DotFlow()
workflow.task.add(step=task_foo, group_name="foo")
workflow.task.add(step=task_bar, group_name="bar")
workflow.start()
return workflow
if __name__ == "__main__":
main()
👀 Full file preview
from time import sleep
from dotflow import DotFlow, action
@action
def task_foo():
sleep(5)
print("2: Task Foo")
return "ok"
@action
def task_bar():
sleep(2)
print("1: Task Bar")
return "ok"
def main():
workflow = DotFlow()
workflow.task.add(step=task_foo, group_name="foo")
workflow.task.add(step=task_bar, group_name="bar")
workflow.start()
return workflow
if __name__ == "__main__":
main()
Sequential Group Mode¶
from time import sleep
from dotflow import DotFlow, action
@action
def task_foo(initial_context):
sleep(2)
value = initial_context.storage
return value * value * value
@action
def task_bar(initial_context):
sleep(1)
value = initial_context.storage
return value * value * value
def main():
workflow = DotFlow()
workflow.task.add(step=task_foo, initial_context=10, group_name="foo")
workflow.task.add(step=task_bar, initial_context=10, group_name="bar")
workflow.start(mode="sequential")
return workflow
if __name__ == "__main__":
main()
👀 Full file preview
from time import sleep
from dotflow import DotFlow, action
@action
def task_foo(initial_context):
sleep(2)
value = initial_context.storage
return value * value * value
@action
def task_bar(initial_context):
sleep(1)
value = initial_context.storage
return value * value * value
def main():
workflow = DotFlow()
workflow.task.add(step=task_foo, initial_context=10, group_name="foo")
workflow.task.add(step=task_bar, initial_context=10, group_name="bar")
workflow.start(mode="sequential")
return workflow
if __name__ == "__main__":
main()