prefect

How do I get the name of a flow from within a task in Prefect?

I'm not sure if this is possible, since flow names are assigned later when a flow is actually run (aka, "creepy-lemur" or whatnot), but I'd like to define a Prefect task within a flow and have that task collect the name of the flow that ran it, so I can insert it into a database table. Has anyone figured out how to do this?

Solution

  • You can get the flow run name and ID from the context:

    import prefect
from prefect import task, flow


@task
def print_task_context():
    print("Task run context:")
    print(prefect.context.get_run_context().task_run.dict())


@flow
def main_flow():
    print_task_context()
    print("Flow run context:")
    print(prefect.context.get_run_context().flow_run.dict())


if __name__ == "__main__":
    main_flow()

    Here are more resources on Prefect Discourse about setting custom run names: