I am using the
create_flow_run_from_deployment
/
run_deployment
APIs to compose my deployments on http request. E.g. I get a payload defining a set of flows to chain together, that triggers a parent container flow, and that container flow does the orchestration of tying things together.
The way I'm thinking of running it, because there are dependencies, is to have the
run_deployment
code wrapped in a task like this :
@task
async def run_deployment(depl_id: str, parameters: dict):
async with prefect.context(**prefect.context.run_params):
async with prefect.Client() as client:
run = await client.create_flow_run_from_deployment(deployment_id=depl_id, parameters=parameters)
run_state = await run.get_state()
return run_state.result
and then in an orchestration flow I think I want to do something like this where I create these tasks for each deployment.
@flow
async def container_flow(flow_graph: dict):
results = {}
ordered_flows = order(flow_graph)
# create tasks
tasks = {name: run_deployment.map(deployment_id=flow_params['deployment_id'], parameters=flow_params['inputs'])
for name, flow_params in flow_graph.items()}
# set dependencies
for flow_name, flow_params in flow_graph.items():
for dependency in flow_params.get('dependencies', []):
tasks[flow_name].set_upstream(tasks[dependency], flow=True)
# run tasks concurrently
flow_results = await prefect.engine.run(tasks, return_tasks=True)
# store results
for flow_name, flow_result in flow_results.items():
results[flow_name] = flow_result.result
return results
