How to access Prefect context values?

Archive
, , ,
1

Prefect 2.0

You may call prefect.context.get_run_context() from a task or a flow to get a TaskRunContext or FlowRunContext object.

The context is immutable and should never be used to pass data. Here is an example showing how to get context-specific information about the task run and flow run:

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()
Example output 
task_run = {
  "id": UUID("af2b80a5-8aaf-449e-a4ba-c60f036afafe"),
  "created": datetime.datetime(
      2022, 1, 14, 13, 11, 32, 364361, tzinfo=datetime.timezone.utc
  ),
  "updated": datetime.datetime(
      2022, 1, 14, 13, 11, 32, 383000, tzinfo=datetime.timezone.utc
  ),
  "name": "print_task_context-6cc57bf3-0",
  "flow_run_id": UUID("e7bf63c3-4d10-4abc-9bc4-69080b297f36"),
  "task_key": "6cc57bf362c4dca11ea2f93bff8c24d3",
  "dynamic_key": "0",
  "cache_key": None,
  "cache_expiration": None,
  "task_version": None,
  "empirical_policy": {"max_retries": 0, "retry_delay_seconds": 0.0},
  "tags": [],
  "state_id": UUID("cbdc71f8-2d86-4cb2-8610-23dd049854fb"),
  "task_inputs": {},
  "state_type": StateType.PENDING,
  "run_count": 0,
  "expected_start_time": datetime.datetime(
      2022, 1, 14, 13, 11, 32, 358365, tzinfo=datetime.timezone.utc
  ),
  "next_scheduled_start_time": None,
  "start_time": None,
  "end_time": None,
  "total_run_time": datetime.timedelta(0),
  "estimated_run_time": datetime.timedelta(0),
  "estimated_start_time_delta": datetime.timedelta(microseconds=28085),
  "state": {
      "id": UUID("cbdc71f8-2d86-4cb2-8610-23dd049854fb"),
      "type": StateType.PENDING,
      "name": "Pending",
      "timestamp": datetime.datetime(
          2022, 1, 14, 13, 11, 32, 358365, tzinfo=datetime.timezone.utc
      ),
      "message": None,
      "data": None,
      "state_details": {
          "flow_run_id": UUID("e7bf63c3-4d10-4abc-9bc4-69080b297f36"),
          "task_run_id": UUID("af2b80a5-8aaf-449e-a4ba-c60f036afafe"),
          "child_flow_run_id": None,
          "scheduled_time": None,
          "cache_key": None,
          "cache_expiration": None,
      },
  },
}
flow_run = {
  "id": UUID("e7bf63c3-4d10-4abc-9bc4-69080b297f36"),
  "created": datetime.datetime(
      2022, 1, 14, 13, 11, 32, 317433, tzinfo=datetime.timezone.utc
  ),
  "updated": datetime.datetime(
      2022, 1, 14, 13, 11, 32, 326000, tzinfo=datetime.timezone.utc
  ),
  "name": "scarlet-elk",
  "flow_id": UUID("b0ec4ef0-067e-4e20-ae67-ea9f35c05d09"),
  "state_id": UUID("fd9cb1de-a645-49c5-a3bb-ea218974271d"),
  "deployment_id": None,
  "flow_version": "4b7387e4e13756dc757aa91519c74f26",
  "parameters": {},
  "idempotency_key": None,
  "context": {},
  "empirical_policy": {},
  "empirical_config": {},
  "tags": [],
  "parent_task_run_id": None,
  "state_type": StateType.PENDING,
  "run_count": 0,
  "expected_start_time": datetime.datetime(
      2022, 1, 14, 13, 11, 32, 306951, tzinfo=datetime.timezone.utc
  ),
  "next_scheduled_start_time": None,
  "start_time": None,
  "end_time": None,
  "total_run_time": datetime.timedelta(0),
  "estimated_run_time": datetime.timedelta(0),
  "estimated_start_time_delta": datetime.timedelta(microseconds=23131),
  "auto_scheduled": False,
  "flow_runner": {"type": None, "config": None},
  "state": {
      "id": UUID("fd9cb1de-a645-49c5-a3bb-ea218974271d"),
      "type": StateType.PENDING,
      "name": "Pending",
      "timestamp": datetime.datetime(
          2022, 1, 14, 13, 11, 32, 306951, tzinfo=datetime.timezone.utc
      ),
      "message": None,
      "data": None,
      "state_details": {
          "flow_run_id": UUID("e7bf63c3-4d10-4abc-9bc4-69080b297f36"),
          "task_run_id": None,
          "child_flow_run_id": None,
          "scheduled_time": None,
          "cache_key": None,
          "cache_expiration": None,
      },
  },
}

For more information, check the API reference.

Prefect 1.0

Context is a thread-safe way of accessing variables related to the flow run and task run. This documentation page provides more information about the key-value pairs available during task runs.

Here is a copy for reference:

Variable Description
date an actual datetime object representing the current time
today the current date formatted as YYYY-MM-DD
today_nodash the current date formatted as YYYYMMDD
yesterday yesterday’s date formatted as YYYY-MM-DD
yesterday_nodash yesterday’s date formatted as YYYYMMDD
tomorrow tomorrow’s date formatted as YYYY-MM-DD
tomorrow_nodash tomorrow’s date formatted as YYYYMMDD
logger the logger for the current task
config the complete Prefect configuration object that is being used during this run
flow_name the name of the current flow
scheduled_start_time a datetime object representing the scheduled start time for the flow run; falls back to now for unscheduled runs
parameters a dictionary of parameter values for the current flow run
map_index the map index of the current task (if mapped, otherwise None )
task_name the name of the current task
task_full_name the name of the current task, including map index
task_slug the slug of the current task
task_tags the tags on the current task
task_run_count the run count of the task run - typically only interesting for retrying tasks
task_loop_count if the Task utilizes looping, the loop count of the task run
task_run_name the run name of the current task (if provided, otherwise None )
task_loop_result if the Task is looping, the current loop result

Prefect Cloud and Server supply some additional context variables:

Variable Description
flow_id the id of the current flow
flow_run_id the id of the current flow run
flow_run_version the state version of the current flow run
flow_run_name the name of the current flow run
task_id the id of the current task
task_run_id the id of the current task run
task_run_version the state version of the current task run
resume boolean showing if the current task run was manually restarted

Users can also provide values to context at runtime. For more information, see the Context concept doc.

1 Like
2

Is there no way to access the flow_name from the flow run context? Was available in 1.0 and seems pretty intuitive. I only see flow_id and flow_run_name above. What’s the easiest way to derive flow_name from flow_id… seems like it should be in the dict by default and not a two step process.

3

Similarly… when you are inside a task you can get the flow_run_id but not the flow_run_name. You need the same two step process here to get the flow_run_name from the inside a task.