https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> Hey, can I access the current flow ru...
# marvin-ai
d
@Marvin Hey, can I access the current flow run id being executed from the task like flow_run_id = get_run_context().flow_run.id ?
m
✅ thought for 97.9 seconds
Not with 
get_run_context().flow_run.id
inside a task. In Prefect 3.x: - 
get_run_context()
inside a flow returns a flow context, so 
ctx.flow_run.id
works there. - 
get_run_context()
inside a task returns a task context, so there is no 
ctx.flow_run
. So this will fail in a task:
Copy code
python
from prefect.context import get_run_context

ctx = get_run_context()
flow_run_id = ctx.flow_run.id  # won't work in a task
Use one of these instead from a task:
Copy code
python
from prefect import task
from prefect.runtime import flow_run

@task
def my_task():
    print(flow_run.id)
Or via the task run context:
Copy code
python
from prefect import task
from prefect.context import get_run_context

@task
def my_task():
    ctx = get_run_context()
    print(ctx.task_run.flow_run_id)
Recommended: 
prefect.runtime.flow_run.id
Example:
Copy code
python
from prefect import flow, task
from prefect.runtime import flow_run

@task
def my_task():
    print(f"Task can see flow run id: {flow_run.id}")

@flow
def my_flow():
    print(f"Flow run id: {flow_run.id}")
    my_task()

my_flow()
So the short answer is: - In a flow: 
get_run_context().flow_run.id
✅ - In a task: use 
flow_run.id
or 
get_run_context().task_run.flow_run_id
✅ Docs: Runtime context
d
@Marvin fine! how about accessing the flow name from the task?
m
✅ thought for 8.3 seconds
Yes — from a task, the easiest way is:
Copy code
python
from prefect import task
from prefect.runtime import flow_run

@task
def my_task():
    print(flow_run.name)
A couple notes: - 
flow_run.name
= the run name of the current flow run - if you want the flow's definition name instead, use:
Copy code
python
from prefect import task
from prefect.runtime import flow

@task
def my_task():
    print(flow.name)
So from inside a task: - current flow run name → 
prefect.runtime.flow_run.name
- current flow name → 
prefect.runtime.flow.name
Example:
Copy code
python
from prefect import flow, task
from prefect.runtime import flow, flow_run

@task
def my_task():
    print(f"Flow name: {flow.name}")
    print(f"Flow run name: {flow_run.name}")

@flow(name="daily-etl")
def my_flow():
    my_task()

my_flow()
Docs: Runtime context
d
@Marvin shouldn't I use the context in this case? Is it right like this?
Copy code
task_name = get_run_context().task.name
    flow_name = flow.name
m
✅ thought for 10.8 seconds
Yes — that is basically right. Inside a task in Prefect 3.x:
Copy code
python
from prefect.context import get_run_context
from prefect.runtime import flow

ctx = get_run_context()
task_name = ctx.task.name
flow_name = flow.name
That’s a good pattern. Quick rule of thumb: - use 
get_run_context()
for task-specific or flow-specific context objects - 
ctx.task.name
- 
ctx.task_run.id
- use 
prefect.runtime
for runtime metadata that should work cleanly across contexts - 
flow.name
- 
flow_run.id
- 
flow_run.name
So for your example:
Copy code
python
task_name = get_run_context().task.name
flow_name = flow.name
Yes, that’s correct. One small nuance: - 
ctx.task.name
is the task definition name - 
flow.name
is the flow definition name - if you wanted the flow run name, use 
flow_run.name
Example:
Copy code
python
from prefect import task
from prefect.context import get_run_context
from prefect.runtime import flow, flow_run

@task
def my_task():
    ctx = get_run_context()
    task_name = ctx.task.name
    flow_name = flow.name
    flow_run_name = flow_run.name

    print(task_name, flow_name, flow_run_name)
If you want, I can also give you a small cheat sheet for what to access from 
get_run_context()
vs 
prefect.runtime
inside tasks and flows.
d
@Marvin 
from prefect.runtime import flow
- flow is underlined red
m
✅ thought for 77.1 seconds
Yep — the red underline is expected. 
from prefect.runtime import flow
is not a valid import in Prefect 3.x. Use one of these instead: To get the flow's name from inside a task:
Copy code
python
from prefect.runtime import flow_run

flow_name = flow_run.flow_name
Or via context:
Copy code
python
from prefect.context import get_run_context

flow_name = get_run_context().flow.name
So your corrected version would be:
Copy code
python
from prefect.context import get_run_context
from prefect.runtime import flow_run

ctx = get_run_context()
task_name = ctx.task.name
flow_name = flow_run.flow_name
Or, if you want to use context for both:
Copy code
python
from prefect.context import get_run_context

ctx = get_run_context()
task_name = ctx.task.name
flow_name = ctx.flow.name
Recommendation: - 
ctx.task.name
for the task name - 
flow_run.flow_name
for the flow name - 
flow_run.name
for the flow run name Docs: Runtime context
Open in Slack
PreviousNext
Powered by