Python Workflows SDK は、step.do とパラメーター名で依存関係（そのステップが走る前に完了していなければならないほかのステップ）を定義し、宣言的に DAG Workflows を書けます。
from workers import Response, WorkflowEntrypoint, WorkerEntrypoint
class PythonWorkflowStarter(WorkflowEntrypoint):
async def run(self, event, step):
async def await_step(fn):
try:
return await fn()
except TypeError as e:
print(f"Successfully caught {type(e).__name__}: {e}")
await step.sleep('demo sleep', '10 seconds')
@step.do()
async def dep_1():
# does stuff
print('executing dep1')
return 'dep1'
@step.do()
async def dep_2():
# does stuff
print('executing dep2')
return 'dep2'
@step.do(concurrent=True)
async def final_step(dep_1, dep_2):
# does stuff
print(f'{dep_1} {dep_2}')
await await_step(final_step)
class Default(WorkerEntrypoint):
async def fetch(self, request):
await self.env.MY_WORKFLOW.create()
return Response("Hello world!")この例では、dep_1 と dep_2 が並行して走り、その両方に依存する final_step が実行されます。
concurrent=True にすると、依存関係を並行して解決できます。すでに完了した依存はスキップされ、戻り値が再利用されます。
このパターンは、並行して走れる 2 つ以上のステップに依存する、ダイヤモンド型の Workflow に向いています。