Access the Apache Airflow context

The Airflow context is a dictionary containing information about a running DAG and its Airflow environment that can be accessed from a task. One of the most common values to retrieve from the Airflow context is the ti / task_instance keyword, which allows you to access attributes and methods of the taskinstance object.

Other common reasons to access the Airflow context are:

• You want to use DAG-level parameters in your Airflow tasks.
• You want to use the DAG run’s logical date in an Airflow task, for example as part of a file name.
• You want to explicitly push and pull values to XCom with a custom key.

Use this document to learn about the data stored in the Airflow context and how to access it.

Assumed knowledge

To get the most out of this guide, you should have an understanding of:

• Basic Airflow concepts. See Introduction to Apache Airflow.
• Basic Python. See the Python Documentation.
• Airflow operators. See Airflow operators.

Access the Airflow context

The Airflow context is available in all Airflow tasks. You can access information from the context using the following methods:

• Pass the **context argument to the function used in a @task decorated task or PythonOperator.
• Pass the context argument to a @asset decorated function. See Assets and data-aware scheduling for more information.
• Use Jinja templating in traditional Airflow operators.
• Access the context kwarg in the .execute method of any traditional or custom operator.

You cannot access the Airflow context dictionary outside of an Airflow task.

Retrieve the Airflow context using the @task decorator or PythonOperator

To access the Airflow context in a @task decorated task or PythonOperator task, you need to add a **context argument to your task function. This will make the context available as a dictionary in your task.

The following code snippets show how to print out the full context dictionary from a task:

1
# from airflow.sdk import task
2
from pprint import pprint
3
4
@task
5
def print_context(**context):
6
    pprint(context)

1
# from airflow.providers.standard.operators.python import PythonOperator
2
from pprint import pprint
3
4
def print_context_func(**context):
5
    pprint(context)
6
7
print_context = PythonOperator(
8
    task_id="print_context",
9
    python_callable=print_context_func,
10
)

Retrieve the Airflow context using Jinja templating

Many elements of the Airflow context can be accessed by using Jinja templating. You can get the list of all parameters that allow templates for any operator by printing out its .template_fields attribute.

For example, you can access a DAG run’s logical date in the format YYYY-MM-DD by using the template {{ ds }} in the bash_command parameter of the BashOperator.

1
# from airflow.providers.standard.operators.bash import BashOperator
2
3
print_logical_date = BashOperator(
4
    task_id="print_logical_date",
5
    bash_command="echo {{ ds }}",
6
)

It is also common to use Jinja templating to access XCom values in the parameter of a traditional task. In the code snippet below, the first task return_greeting will push the string “Hello” to XCom, and the second task greet_friend will use a Jinja template to pull that value from the ti (task instance) object of the Airflow context and print Hello friend! :) into the logs.

1
# from airflow.providers.standard.operators.bash import BashOperator
2
# from airflow.sdk import task
3
4
@task 
5
def return_greeting():
6
    return "Hello"
7
8
greet_friend = BashOperator(
9
    task_id="greet_friend",
10
    bash_command="echo '{{ ti.xcom_pull(task_ids='return_greeting') }} friend! :)'",
11
)
12
13
return_greeting() >> greet_friend

Find an up to date list of all available templates in the Airflow documentation. Learn more about using XComs to pass data between Airflow tasks in Pass data between tasks.

Retrieve the Airflow context using custom operators

In a traditional operator, the Airflow context is always passed to the .execute method using the context keyword argument. If you write a custom operator, you have to include a context kwarg in the execute method as shown in the following custom operator example.

1
from airflow.sdk.bases.operator import BaseOperator
2
3
class PrintDAGIDOperator(BaseOperator):
4
    def __init__(self, *args, **kwargs):
5
        super().__init__(*args, **kwargs)
6
7
    def execute(self, context):
8
        print(context["dag"].dag_id)

Common Airflow context values

This section gives an overview of the most commonly used keys in the Airflow context dictionary. To see an up-to-date list of all keys and their types, view the Airflow source code.

ti / task_instance

The ti or task_instance key contains the TaskInstance object. The most commonly used attributes are .xcom_pull and .xcom_push, which allow you to push and pull XComs.

The following DAG shows an example of using context["ti"].xcom_push(...) and context["ti"].xcom_pull(...) to explicitly pass data between tasks.

1
from pendulum import datetime
2
from airflow.decorators import dag, task
3
4
5
@dag(
6
    start_date=datetime(2023, 6, 1),
7
    schedule=None,
8
    catchup=False,
9
)
10
def context_and_xcom():
11
    @task
12
    def upstream_task(**context):
13
        context["ti"].xcom_push(key="my_explicitly_pushed_xcom", value=23)
14
        return 19
15
16
    @task
17
    def downstream_task(passed_num, **context):
18
        returned_num = context["ti"].xcom_pull(
19
            task_ids="upstream_task", key="return_value"
20
        )
21
        explicit_num = context["ti"].xcom_pull(
22
            task_ids="upstream_task", key="my_explicitly_pushed_xcom"
23
        )
24
25
        print("Returned Num: ", returned_num)
26
        print("Passed Num: ", passed_num)
27
        print("Explicit Num: ", explicit_num)
28
29
    downstream_task(upstream_task())
30
31
32
context_and_xcom()

The downstream_task will print the following information to the logs:

[2023-06-16, 13:14:11 UTC] {logging_mixin.py:149} INFO - Returned Num:  19
[2023-06-16, 13:14:11 UTC] {logging_mixin.py:149} INFO - Passed Num:  19
[2023-06-16, 13:14:11 UTC] {logging_mixin.py:149} INFO - Explicit Num:  23

Scheduling keys

One of the most common reasons to access the Airflow context in your tasks is to retrieve information about the scheduling of their DAG. A common pattern is to use the timestamp of the logical date in names of files written from a DAG to create a unique file for each DAG run.

The task below creates a new text file in the include folder for each DAG run with the timestamp in the filename in the format YYYY-MM-DDTHH:MM:SS+00:00. Refer to Templates reference for an up to date list of time related keys in the context, and Jinja templating for more information on how to pass these values to templateable parameters of traditional operators.

1
# from airflow.sdk import task
2
3
@task
4
def write_file_with_ts(**context):
5
    ts = context["ts"]
6
    with open(f"include/{ts}_hello.txt", "a") as f:
7
        f.write("Hello, World!")

dag_run

The dag_run key contains the DAG run object. A commonly used attribute of the DAG run object is run_type, which indicates how the DAG was triggered.

1
# from airflow.sdk import task
2
3
@task
4
def print_dagrun_info(**context):
5
    print(context["dag_run"].run_type)

params

The params key contains a dictionary of all DAG- and task-level params that were passed to a specific task instance. Individual params can be accessed using their respective key.

1
# from airflow.sdk import task
2
3
@task
4
def print_param(**context):
5
    print(context["params"]["my_favorite_param"])

Learn more about params in the Airflow params guide.

var

The var key contains all Airflow variables of your Airflow instance. Airflow variables are key-value pairs that are commonly used to store instance-level information that rarely changes.

1
# from airflow.sdk import task
2
3
@task
4
def get_var_from_context(**context):
5
    print(context["var"]["value"].get("my_regular_var"))
6
    print(context["var"]["json"].get("my_json_var")["num2"])

Context parameters relating to timestamps

Your dag run type, i.e. scheduled vs asset-triggered can determine which timestamp keys are available in the context. The following code snippet contains a task that prints out the full list of context keys available, as well as all keys relating to scheduling timestamps.

1
from airflow.sdk import dag, task, task_group, Asset
2
3
@dag
4
def my_context_dag():
5
    @task 
6
    def print_context_keys(**context):
7
        print("All context keys: ", context.keys())
8
        print("--------------")
9
        print("DAG run details relating to timestamps:")
10
        print("run_id from the dag_run key: ", context["dag_run"].run_id)
11
        print("logical_date from the dag_run key: ", context["dag_run"].logical_date)
12
        print("data_interval_start from the dag_run key: ", context["dag_run"].data_interval_start)
13
        print("data_interval_end from the dag_run key: ", context["dag_run"].data_interval_end)
14
        print("run_after from the dag_run key: ", context["dag_run"].run_after)
15
        print("start_date from the dag_run key: ", context["dag_run"].start_date) 
16
        print("end_date from the dag_run key: ", context["dag_run"].end_date)
17
        print("--------------")
18
        print("Top-level context keys realating to timstamps:")
19
        print("prev_start_date_success: ", context["prev_start_date_success"])
20
        print("prev_end_date_success: ", context["prev_end_date_success"])
21
22
        # The keys below are only available in a scheduled run, or manual run with a logical date provided
23
        # in an asset triggered run or a manual run with no logical date provided, these keys will be MISSING
24
        # causing a key error if used!
25
        print("logical_date: ", context["logical_date"])
26
        print("ds: ", context["ds"])
27
        print("ds_nodash: ", context["ds_nodash"])
28
        print("ts: ", context["ts"])
29
        print("ts_nodash: ", context["ts_nodash"])
30
        print("data_interval_start: ", context["data_interval_start"])        
31
        print("data_interval_end: ", context["data_interval_end"])
32
        print("previous_data_interval_start_success: ", context["prev_data_interval_start_success"])
33
        print("previous_data_interval_end_success: ", context["prev_data_interval_end_success"])
34
35
    print_context_keys()
36
37
my_context_dag()

Note that if your DAG is triggered by an asset or if you created a manual / API triggered run and set the logical date explicitly to None, the following keys will be missing from the context dictionary and trying to access them will raise a KeyError:

• logical_date
• ds
• ds_nodash
• ts
• ts_nodash
• data_interval_start
• data_interval_end
• previous_data_interval_start_success
• previous_data_interval_end_success