Hammad Ahmed
01/05/2022, 4:26 PMKevin Kho
create_flow_run
just creates a flow with schedule = now. I think in order to do this, you would need to pass it in as a parameter to the child flow. This experience will be improved in Orion thoughHammad Ahmed
01/05/2022, 4:31 PMAnna Geller
prefect.context.flow_id
If you meant flow run id, then:
prefect.context.flow_run_id
and the child flow run ID is the result of the create_flow_run taskKevin Kho
Anna Geller
import prefect
from prefect import Flow, task
from prefect.tasks.prefect import create_flow_run, wait_for_flow_run
@task
def get_parent_flow_run_id():
return prefect.context.flow_run_id
with Flow("parametrized_child_flow_run_from_parent_flow") as flow:
parent_flow_run_id = get_parent_flow_run_id()
child_flow_run_id = create_flow_run(
flow_name="parametrized_flow", project_name="community", parameters=dict(parent_flow_run_id=parent_flow_run_id)
)
wait_for_flow_run(child_flow_run_id, stream_logs=True, raise_final_state=True)