source_dag_run does not exist on AssetEventDagRunReferenceResult

[jrderuiter]

Apache Airflow version

3.0.2

If "Other Airflow 2 version" selected, which one?

No response

What happened?

The docs state that you can use the source_dag_run attribute to fetch information about a triggering DAG run when using asset-aware scheduling. However, when I try to do so this attribute doesn't seem to exist in Airflow 3, resulting in an error:

[2025-07-06, 10:52:29] ERROR - Task failed with exception: source="task"
AttributeError: 'AssetEventDagRunReferenceResult' object has no attribute 'source_dag_run'

Drilling into the AssetEventDagRunReferenceResult class I also don't see this attribute being defined (only DAG and run IDs).

What you think should happen instead?

The given example in the docs should work. The source_dag_run attribute should be available and provide access to the source DagRun instance.

How to reproduce

Code to reproduce:

import pendulum
from airflow.providers.standard.operators.python import PythonOperator
from airflow.sdk import DAG, Asset
from airflow.sdk.execution_time.comms import AssetEventDagRunReferenceResult
from airflow.sdk.execution_time.context import TriggeringAssetEventsAccessor

dataset = Asset("/data/events")

with DAG(dag_id="example_producer", start_date=pendulum.today()):
    PythonOperator(
        task_id="produce",
        python_callable=lambda **context: print("Produce!"),
        outlets=[dataset],
    )


def _print_context(triggering_asset_events: TriggeringAssetEventsAccessor):
    result: AssetEventDagRunReferenceResult = triggering_asset_events[dataset][0]
    print(result.source_dag_run) # Errors here


with DAG(dag_id="example_consumer", schedule=[dataset], start_date=pendulum.today()):
    PythonOperator(
        task_id="print_context",
        python_callable=_print_context,
        outlets=[dataset],
    )

This uses a Python function to look at the context, but accessing the source_dag_run via Jinja templating (as shown in the docs) gives the same result.

Operating System

Docker

Versions of Apache Airflow Providers

No response

Deployment

Docker-Compose

Deployment details

No response

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[tirkarthi]

cc: @uranusjr @Lee-W

[charlie-liner]

Interesting issue! I am looking into it

[Lee-W]

Interesting issue! I am looking into it

Sounds good. I've not yet have the bandwidth to look into it. Thanks for helping out! Just assigned it to you.

[kyungjunleeme]

@charlie-liner @kgw7401

#41348
this pr is large refactoring pr

TaskInstance (source) → AssetEvent is recorded → source_dag_id and source_run_id are stored
                                           ↓
                             source_dag_run relationship allows tracing back to the DagRun

class AssetEvent(Base):
    """
    A table to store assets events.

    :param asset_id: reference to AssetModel record
    :param extra: JSON field for arbitrary extra info
    :param source_task_id: the task_id of the TI which updated the asset
    :param source_dag_id: the dag_id of the TI which updated the asset
    :param source_run_id: the run_id of the TI which updated the asset
    :param source_map_index: the map_index of the TI which updated the asset
    :param timestamp: the time the event was logged

    We use relationships instead of foreign keys so that asset events are not deleted even
    if the foreign key object is.
    """

    id = Column(Integer, primary_key=True, autoincrement=True)
    asset_id = Column(Integer, nullable=False)
    extra = Column(sqlalchemy_jsonfield.JSONField(json=json), nullable=False, default={})
    source_task_id = Column(StringID(), nullable=True)
    source_dag_id = Column(StringID(), nullable=True)
    source_run_id = Column(StringID(), nullable=True)
    source_map_index = Column(Integer, nullable=True, server_default=text("-1"))
    timestamp = Column(UtcDateTime, default=timezone.utcnow, nullable=False)

    __tablename__ = "asset_event"
    __table_args__ = (
        Index("idx_asset_id_timestamp", asset_id, timestamp),
        {"sqlite_autoincrement": True},  # ensures PK values not reused
    )

    created_dagruns = relationship(
        "DagRun",
        secondary=association_table,
        backref="consumed_asset_events",
    )

    source_aliases = relationship(
        "AssetAliasModel",
        secondary=asset_alias_asset_event_association_table,
        back_populates="asset_events",
    )

    source_task_instance = relationship(
        "TaskInstance",
        primaryjoin="""and_(
            AssetEvent.source_dag_id == foreign(TaskInstance.dag_id),
            AssetEvent.source_run_id == foreign(TaskInstance.run_id),
            AssetEvent.source_task_id == foreign(TaskInstance.task_id),
            AssetEvent.source_map_index == foreign(TaskInstance.map_index),
        )""",
        viewonly=True,
        lazy="select",
        uselist=False,
    )
    source_dag_run = relationship(
        "DagRun",
        primaryjoin="""and_(
            AssetEvent.source_dag_id == foreign(DagRun.dag_id),
            AssetEvent.source_run_id == foreign(DagRun.run_id),
        )""",
        viewonly=True,
        lazy="select",
        uselist=False,
    )
    asset = relationship(
        AssetModel,
        primaryjoin="AssetEvent.asset_id == foreign(AssetModel.id)",
        viewonly=True,
        lazy="select",
        uselist=False,
    )

it can be helpful.