> ## Documentation Index
> Fetch the complete documentation index at: https://astronomer.io/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# 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](#ti-/-task_instance), which allows you to access attributes and methods of the [`taskinstance` object](https://airflow.apache.org/docs/apache-airflow/stable/_api/airflow/models/taskinstance/index.html).

Other common reasons to access the Airflow context are:

* You want to use [DAG-level parameters](/docs/learn/airflow-params) in your Airflow tasks.
* You want to use the DAG run's [logical date](/docs/learn/scheduling-in-airflow#dag-run-timestamps) in an Airflow task, for example as part of a file name.
* You want to explicitly push and pull values to [XCom](/docs/learn/airflow-passing-data-between-tasks#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](/docs/learn/intro-to-airflow).
* Basic Python. See the [Python Documentation](https://docs.python.org/3/tutorial/index.html).
* Airflow operators. See [Airflow operators](/docs/learn/what-is-an-operator).

## 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](/docs/learn/airflow-decorators) or [`PythonOperator`](https://airflow.apache.org/registry/providers/standard#standard-python-PythonOperator).
* Pass the `context` argument to a `@asset` decorated function. See [Assets and data-aware scheduling](/docs/learn/airflow-datasets) for more information.
* Use [Jinja templating](/docs/learn/templating) in traditional Airflow operators.
* Access the context `kwarg` in the `.execute` method of any traditional or custom operator.

You can't 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:

<details open>
  <summary>TaskFlow</summary>

  ```python wrap theme={null}
  # from airflow.sdk import task
  from pprint import pprint

  @task
  def print_context(**context):
      pprint(context)
  ```
</details>

<details>
  <summary>Traditional</summary>

  ```python wrap theme={null}
  # from airflow.providers.standard.operators.python import PythonOperator
  from pprint import pprint

  def print_context_func(**context):
      pprint(context)

  print_context = PythonOperator(
      task_id="print_context",
      python_callable=print_context_func,
  )
  ```
</details>

### Retrieve the Airflow context using Jinja templating

Many elements of the Airflow context can be accessed by using [Jinja templating](/docs/learn/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`.

```python wrap theme={null}
# from airflow.providers.standard.operators.bash import BashOperator

print_logical_date = BashOperator(
    task_id="print_logical_date",
    bash_command="echo {{ ds }}",
)
```

It is also common to use Jinja templating to access [XCom](/docs/learn/airflow-passing-data-between-tasks#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.

```python wrap theme={null}
# from airflow.providers.standard.operators.bash import BashOperator
# from airflow.sdk import task

@task 
def return_greeting():
    return "Hello"

greet_friend = BashOperator(
    task_id="greet_friend",
    bash_command="echo '{{ ti.xcom_pull(task_ids='return_greeting') }} friend! :)'",
)

return_greeting() >> greet_friend
```

Find an up to date list of all available templates in the [Airflow documentation](https://airflow.apache.org/docs/apache-airflow/stable/templates-ref.html). Learn more about using XComs to pass data between Airflow tasks in [Pass data between tasks](/docs/learn/airflow-passing-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](/docs/learn/airflow-importing-custom-hooks-operators), you have to include a `context` kwarg in the `execute` method as shown in the following custom operator example.

```python wrap theme={null}
from airflow.sdk.bases.operator import BaseOperator

class PrintDAGIDOperator(BaseOperator):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def execute(self, context):
        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](https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/utils/context.py).

### ti / `task_instance`

The `ti` or `task_instance` key contains the [TaskInstance object](https://airflow.apache.org/docs/apache-airflow/stable/_api/airflow/models/taskinstance/index.html). The most commonly used attributes are `.xcom_pull` and `.xcom_push`, which allow you to push and pull [XComs](/docs/learn/airflow-passing-data-between-tasks).

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

```python expandable wrap theme={null}
from pendulum import datetime
from airflow.decorators import dag, task


@dag(
    start_date=datetime(2023, 6, 1),
    schedule=None,
    catchup=False,
)
def context_and_xcom():
    @task
    def upstream_task(**context):
        context["ti"].xcom_push(key="my_explicitly_pushed_xcom", value=23)
        return 19

    @task
    def downstream_task(passed_num, **context):
        returned_num = context["ti"].xcom_pull(
            task_ids="upstream_task", key="return_value"
        )
        explicit_num = context["ti"].xcom_pull(
            task_ids="upstream_task", key="my_explicitly_pushed_xcom"
        )

        print("Returned Num: ", returned_num)
        print("Passed Num: ", passed_num)
        print("Explicit Num: ", explicit_num)

    downstream_task(upstream_task())


context_and_xcom()
```

The `downstream_task` will print the following information to the logs:

```text wrap theme={null}
[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](https://airflow.apache.org/docs/apache-airflow/stable/templates-ref.html) for an up to date list of time related keys in the context, and [Jinja templating](/docs/learn/templating) for more information on how to pass these values to templateable parameters of traditional operators.

```python wrap theme={null}
# from airflow.sdk import task

@task
def write_file_with_ts(**context):
    ts = context["ts"]
    with open(f"include/{ts}_hello.txt", "a") as f:
        f.write("Hello, World!")
```

### `dag_run`

The `dag_run` key contains the [DAG run object](https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/models/dagrun.py). A commonly used attribute of the DAG run object is `run_type`, which indicates how the DAG was triggered.

```python wrap theme={null}
# from airflow.sdk import task

@task
def print_dagrun_info(**context):
    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.

```python wrap theme={null}
# from airflow.sdk import task

@task
def print_param(**context):
    print(context["params"]["my_favorite_param"])
```

Learn more about params in the [Airflow params guide](/docs/learn/airflow-params).

### var

The `var` key contains all Airflow variables of your Airflow instance. [Airflow variables](/docs/learn/airflow-variables) are key-value pairs that are commonly used to store instance-level information that rarely changes.

```python wrap theme={null}
# from airflow.sdk import task

@task
def get_var_from_context(**context):
    print(context["var"]["value"].get("my_regular_var"))
    print(context["var"]["json"].get("my_json_var")["num2"])
```

## Context parameters relating to timestamps

Your dag run type, that is 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.

```python expandable wrap theme={null}
from airflow.sdk import dag, task, task_group, Asset

@dag
def my_context_dag():
    @task 
    def print_context_keys(**context):
        print("All context keys: ", context.keys())
        print("--------------")
        print("DAG run details relating to timestamps:")
        print("run_id from the dag_run key: ", context["dag_run"].run_id)
        print("logical_date from the dag_run key: ", context["dag_run"].logical_date)
        print("data_interval_start from the dag_run key: ", context["dag_run"].data_interval_start)
        print("data_interval_end from the dag_run key: ", context["dag_run"].data_interval_end)
        print("run_after from the dag_run key: ", context["dag_run"].run_after)
        print("start_date from the dag_run key: ", context["dag_run"].start_date) 
        print("end_date from the dag_run key: ", context["dag_run"].end_date)
        print("--------------")
        print("Top-level context keys relating to timestamps:")
        print("prev_start_date_success: ", context["prev_start_date_success"])
        print("prev_end_date_success: ", context["prev_end_date_success"])

        # The keys below are only available in a scheduled run, or manual run with a logical date provided
        # in an asset triggered run or a manual run with no logical date provided, these keys will be MISSING
        # causing a key error if used!
        print("logical_date: ", context["logical_date"])
        print("ds: ", context["ds"])
        print("ds_nodash: ", context["ds_nodash"])
        print("ts: ", context["ts"])
        print("ts_nodash: ", context["ts_nodash"])
        print("data_interval_start: ", context["data_interval_start"])        
        print("data_interval_end: ", context["data_interval_end"])
        print("previous_data_interval_start_success: ", context["prev_data_interval_start_success"])
        print("previous_data_interval_end_success: ", context["prev_data_interval_end_success"])

    print_context_keys()

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`