jack
04/05/2023, 6:36 PMflow_run.name
from inside a task?
From inside a flow this works:
from prefect.context import get_run_context
get_run_context().flow_run.name
but inside a task, it throws
AttributeError: 'TaskRunContext' object has no attribute 'flow_run'
alex
04/05/2023, 8:21 PMdef get_flow_run_id() -> str:
run_context = get_run_context()
if isinstance(run_context, TaskRunContext):
flow_run_id = run_context.task_run.dict().get("flow_run_id")
else:
flow_run_id = run_context.flow_run.dict().get("id")
return str(flow_run_id)
jack
04/05/2023, 10:54 PMflow_run_name
from the flow_run_id
?alex
04/06/2023, 2:25 PMjack
04/06/2023, 4:41 PMimport prefect.runtime
prefect.runtime.flow_run.name