You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "jrderuiter (via GitHub)" <gi...@apache.org> on 2025/07/06 09:03:20 UTC
[I] source_dag_run does not exist on AssetEventDagRunReferenceResult [airflow]
jrderuiter opened a new issue, #52932:
URL: https://github.com/apache/airflow/issues/52932
### 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 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'
```
### 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)
with DAG(dag_id="example_consumer", schedule=[dataset], start_date=pendulum.today()):
PythonOperator(
task_id="print_context",
python_callable=_print_context,
outlets=[dataset],
)
```
### 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
- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] source_dag_run does not exist on AssetEventDagRunReferenceResult [airflow]
Posted by "tirkarthi (via GitHub)" <gi...@apache.org>.
tirkarthi commented on issue #52932:
URL: https://github.com/apache/airflow/issues/52932#issuecomment-3045198753
cc: @uranusjr @Lee-W
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] source_dag_run does not exist on AssetEventDagRunReferenceResult [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr closed issue #52932: source_dag_run does not exist on AssetEventDagRunReferenceResult
URL: https://github.com/apache/airflow/issues/52932
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] source_dag_run does not exist on AssetEventDagRunReferenceResult [airflow]
Posted by "charlie-liner (via GitHub)" <gi...@apache.org>.
charlie-liner commented on issue #52932:
URL: https://github.com/apache/airflow/issues/52932#issuecomment-3050746130
Interesting issue! I am looking into it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] source_dag_run does not exist on AssetEventDagRunReferenceResult [airflow]
Posted by "kyungjunleeme (via GitHub)" <gi...@apache.org>.
kyungjunleeme commented on issue #52932:
URL: https://github.com/apache/airflow/issues/52932#issuecomment-3069588649
@charlie-liner @kgw7401
https://github.com/apache/airflow/pull/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
```
```python
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] source_dag_run does not exist on AssetEventDagRunReferenceResult [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on issue #52932:
URL: https://github.com/apache/airflow/issues/52932#issuecomment-3050759353
> 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org