source_dag_run does not exist on AssetEventDagRunReferenceResult

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

@uranusjr

cc: @uranusjr @Lee-W

Interesting issue! I am looking into it

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.

@charlie-liner

@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.

Navigation Menu

Repository navigation

source_dag_run does not exist on AssetEventDagRunReferenceResult #52932

Apache Airflow version

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

What happened?

What you think should happen instead?

How to reproduce

Operating System

Versions of Apache Airflow Providers

Deployment

Deployment details

Anything else?

Are you willing to submit PR?

Code of Conduct

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

source_dag_run does not exist on AssetEventDagRunReferenceResult #52932

Description

Apache Airflow version

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

What happened?

What you think should happen instead?

How to reproduce

Operating System

Versions of Apache Airflow Providers

Deployment

Deployment details

Anything else?

Are you willing to submit PR?

Code of Conduct

Activity

tirkarthi commented on Jul 7, 2025

charlie-liner commented on Jul 9, 2025

Lee-W commented on Jul 9, 2025

kyungjunleeme commented on Jul 14, 2025

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions