DAG callback context does not have dag_run set in Airflow 3 but it is set in Airflow 2.

[benitezfede]

Hi,
In the previous versions of airflow I was able to get the dag_run information from the on_success_callback context when a DAG succeeded callback was triggered.

Now, in Airflow 3, I see that the dag_run is no longer populated in the callback context. In my opinion this is a bug, also it is inconsistent with the tasks callbacks as IT IS populated there.

I read the documentation about callbacks here and it points to this the context template variables which have the dag_run. Also the code here shows the dag_run so I cannot understand why it is not being populated.

The fix seems to be a matter of adding dag_run: self, here.

In order to reproduce the issue, I created this DAG which prints the keys of the context for DAG and tasks callbacks:

from datetime import datetime, timedelta
import logging
from airflow import DAG
from airflow.operators.python import PythonOperator

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Debug callback functions
def debug_task_success_callback(context):
    """Debug callback for task success - prints all context information"""
    print("=" * 80)
    print("TASK SUCCESS CALLBACK DEBUG")
    print("=" * 80)
    print(f"Context keys: {list(context.keys())}")
    print(f"Context type: {type(context)}")
    
    # Print each context key and its value
    for key, value in context.items():
        print(f"  {key}: {type(value)} = {value}")
        if hasattr(value, '__dict__'):
            print(f"    Attributes: {[attr for attr in dir(value) if not attr.startswith('_')]}")
    
    # Specifically check for dag_run
    if 'dag_run' in context:
        print(f"\nDAG_RUN FOUND:")
        print(f"  Type: {type(context['dag_run'])}")
        print(f"  Attributes: {[attr for attr in dir(context['dag_run']) if not attr.startswith('_')]}")
        print(f"  Dict: {context['dag_run'].__dict__}")
    else:
        print(f"\nDAG_RUN NOT FOUND in context!")
    
    print("=" * 80)

def debug_dag_success_callback(context):
    """Debug callback for DAG success - prints all context information"""
    print("=" * 80)
    print("DAG SUCCESS CALLBACK DEBUG")
    print("=" * 80)
    print(f"Context keys: {list(context.keys())}")
    print(f"Context type: {type(context)}")
    
    # Print each context key and its value
    for key, value in context.items():
        print(f"  {key}: {type(value)} = {value}")
        if hasattr(value, '__dict__'):
            print(f"    Attributes: {[attr for attr in dir(value) if not attr.startswith('_')]}")
    
    # Specifically check for dag_run
    if 'dag_run' in context:
        print(f"\nDAG_RUN FOUND:")
        print(f"  Type: {type(context['dag_run'])}")
        print(f"  Attributes: {[attr for attr in dir(context['dag_run']) if not attr.startswith('_')]}")
        print(f"  Dict: {context['dag_run'].__dict__}")
    else:
        print(f"\nDAG_RUN NOT FOUND in context!")
    
    print("=" * 80)

def simple_task():
    """Simple task that just prints a message"""
    print("Hello from simple task!")
    return "task completed"

# Define the DAG
default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2023, 1, 1),
    'email': ['airflow@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(seconds=10),
    # Add task-level callback
    'on_success_callback': debug_task_success_callback,
}

dag = DAG(
    'debug_context_dag',
    default_args=default_args,
    description='A simple DAG to debug context in Airflow 3.0',
    schedule=timedelta(days=1),
    catchup=False,
    tags=['debug', 'context'],
    # Add DAG-level callback
    on_success_callback=debug_dag_success_callback,
)

# Define the task
task1 = PythonOperator(
    task_id='simple_task',
    python_callable=simple_task,
    dag=dag,
)

# Set task dependencies (none needed for single task)
task1 

The output of running this DAG shows that for the DAG callback the dag_run is not present:

DAG_RUN NOT FOUND in context!

The context for DAG callbacks has only these keys:

"Context keys: ['dag', 'run_id', 'reason']"

For the task callback we see the following keys:

Context keys: ['dag', 'inlets', 'map_index_template', 'outlets', 'run_id', 'task', 'task_instance', 'ti', 'outlet_events', 'inlet_events', 'macros', 'params', 'var', 'conn', 'dag_run', 'triggering_asset_events', 'task_instance_key_str', 'task_reschedule_count', 'prev_start_date_success', 'prev_end_date_success', 'logical_date', 'ds', 'ds_nodash', 'ts', 'ts_nodash', 'ts_nodash_with_tz', 'data_interval_end', 'data_interval_start', 'prev_data_interval_start_success', 'prev_data_interval_end_success', 'templates_dict', 'exception']

And it has the dag_run as well:

DAG_RUN FOUND:: chan="stdout": source="task"
[2025-07-21, 23:28:17] INFO -   Type: <class 'airflow.sdk.api.datamodels._generated.DagRun'>: chan="stdout": source="task"
[2025-07-21, 23:28:17] INFO -   Attributes: ['clear_number', 'conf', 'construct', 'consumed_asset_events', 'copy', 'dag_id', 'data_interval_end', 'data_interval_start', 'dict', 'end_date', 'from_orm', 'json', 'logical_date', 'model_computed_fields', 'model_config', 'model_construct', 'model_copy', 'model_dump', 'model_dump_json', 'model_extra', 'model_fields', 'model_fields_set', 'model_json_schema', 'model_parametrized_name', 'model_post_init', 'model_rebuild', 'model_validate', 'model_validate_json', 'model_validate_strings', 'parse_file', 'parse_obj', 'parse_raw', 'run_after', 'run_id', 'run_type', 'schema', 'schema_json', 'start_date', 'update_forward_refs', 'validate']

So there is a massive difference in the information loaded in the context for the task callbacks vs the DAG callback. Is this expected or is it a bug? As I said it seems to be a bug.

Thanks,
Fede

[davidsteinar]

Just wanted to say that I am facing the same issue when trying to update my code base from Airflow 2 to 3. I have multiple DAG level callback functions that access the context for posting to Teams channels. I am only getting context object of this type

{'dag': <class 'airflow.sdk.definitions.dag.DAG'>, 'run_id': str, 'reason': str} 

but am unable to find the keys I am interested in logging in the 'dag' object.

[potiuk]

Yes. That's deliberate change described in the changelog - one of the hallmark features of Airflow 3 where we isolated DB access from worker. Previously you could do more things with dag because it was an ORM object representing database DAG table. Now it is just an "object" that is passed through HTTP Task SDK.

You should look at the changelog and see the recommendation and explanations in documentation. Including the advise to use Airflow's REST API to access the information about Dags, DagRuns and other airflow properties. This has been recommended for about 4 years now and the DB access was marked as not part of the publlc API since generally the beginning of Airflow 2 - Airflow 3 now enforces it and does not allow direct database access.

If you feel like some things (from the logical point of view) should be available via task-sdk and you would think that accessing it in your worker without having to use public REST API, I recommend you to open an issue explainig the use pattern you want to have, why you think it is a common pattern that others will want to use, and possibly look at task SDK and propose where and how it could be implemented. We are doing it already when some of the "often used" access methods were commonly used and are missing now - for example #53628 - and we might consider adding it (but it's not like to follow previous access pattern where you could directly access "Database ORM objects) - you need to explain what exactly you need to access (specific logic you want to use it for) and explain why current REST API is not enough (basically what makes you think that the useage is common enough and "task" bound - so that you think it shoudl be built-in).

[davidsteinar]

@potiuk thank you for your reply. In my case I was accessing these objects from the DAG callback context in Airflow 2:

context['data_interval_start']
context["dag_run"].get_task_instances()

how would you access these now in airflow 3 from only the limited context object ?

[potiuk]

REST API

[potiuk]

Or if you have specific use case and can describe why and how you want to access the data from callback - open an issue and I am sure it will be evaluated. But you have to describe your use case in detail.

[potiuk]

Open a new issue and describe it.

[DonHaul]

@benitezfede a workaround I found that works for me is using Notifiers , and pass the extra parameters I need through it. Might also work for your usecase

[mbarugelCA]

I am having a similar issue too. I used context["dag_run"].task_instances within on_failure_callback to report some stats (e.g., list the failed tasks). I also used DagRun.find() to report additional stats on failure, such as how often the DAG has failed over the past X days.

The recommended pattern now is to use the REST API? Does this mean I need to set up API auth to get access to this info? Additionally, does that mean I need to have the API server running to run airflow dags test my_dag and access the DagRun info?

Looking at the available methods in the objects, I found that the DagRun is still accessible with the following code.

dag_run = context["dag"].fetch_dagrun(context["dag"].safe_dag_id, context["run_id"])
tis = dag_run.get_task_instances()

Is this pattern OK to use, or is it expected to go away in future releases of Airflow 3?

[gopidesupavan]

can you please test on the latest RC #54124 there is fix for this #53058

[matthewblock]

I'm running into similar issues - I have been using airflow.models.XXX for type hints, so I could tell which information was available in each object.

from airflow.models.task_instance import TaskInstance

def my_on_success_callback(context):
    ti: TaskInstance = context["task_instance"]
    print(f"{ti=}")

The above now gives me RuntimeError: Direct database access via the ORM is not allowed in Airflow 3.0.

Is there a drop in replacement via airflow.sdk or are we expected to memorize which fields are in which object?

Where can we find what each context contains, for each callback?

[potiuk]

Where can we find what each context contains, for each callback?

Maybe in the Documentation ? Who knows. Also this is an open-source project run by volunteers, and if you need any information missing in docs - you can always look at the source code. Some users also decide to give back and if they find missing or imprecise, or difficult to find documentation, look for it (also in the code) and help the whole community and other users by contributing missing docs and links. Highly recommended as a thank you for the software that they get for free.

BTW. We have 3500 contributors. Many of them users who contributed missing docs and generally - gave back.

[matthewblock]

Where can we find what each context contains, for each callback?

Maybe in the Documentation ? Who knows. Also this is an open-source project run by volunteers, and if you need any information missing in docs - you can always look at the source code. Some users also decide to give back and if they find missing or imprecise, or difficult to find documentation, look for it (also in the code) and help the whole community and other users by contributing missing docs and links. Highly recommended as a thank you for the software that they get for free.

BTW. We have 3500 contributors. Many of them users who contributed missing docs and generally - gave back.

Thanks - I assumed it existed and I missed it.

[kaxil]

It's here: https://airflow.apache.org/docs/task-sdk/stable/api.html#airflow.sdk.Context:~:text=airflow.sdk.Context-,Jinja2%20template%20context%20for%20task%20rendering.,-conn

and

airflow/task-sdk/src/airflow/sdk/definitions/context.py

Lines 40 to 83 in 9ef4f12

	class Context(TypedDict, total=False):
	"""Jinja2 template context for task rendering."""
	conn: Any
	dag: DAG
	dag_run: DagRunProtocol
	data_interval_end: DateTime | None
	data_interval_start: DateTime | None
	outlet_events: OutletEventAccessorsProtocol
	ds: str
	ds_nodash: str
	expanded_ti_count: int | None
	exception: None | str | BaseException
	inlets: list
	inlet_events: InletEventsAccessors
	logical_date: DateTime
	macros: Any
	map_index_template: str | None
	outlets: list
	params: dict[str, Any]
	prev_data_interval_start_success: DateTime | None
	prev_data_interval_end_success: DateTime | None
	prev_start_date_success: DateTime | None
	prev_end_date_success: DateTime | None
	reason: str | None
	run_id: str
	start_date: DateTime
	# TODO: Remove Operator from below once we have MappedOperator to the Task SDK
	# and once we can remove context related code from the Scheduler/models.TaskInstance
	task: BaseOperator | Operator
	task_reschedule_count: int
	task_instance: RuntimeTaskInstanceProtocol
	task_instance_key_str: str
	# `templates_dict` is only set in PythonOperator
	templates_dict: dict[str, Any] | None
	test_mode: bool
	ti: RuntimeTaskInstanceProtocol
	# triggering_asset_events: Mapping[str, Collection[AssetEvent | AssetEventPydantic]]
	triggering_asset_events: Any
	try_number: int | None
	ts: str
	ts_nodash: str
	ts_nodash_with_tz: str
	var: Any