diff --git a/airflow/models/trigger.py b/airflow/models/trigger.py index 5a46fcbda27c2..2e0fe9f7f2bbe 100644 --- a/airflow/models/trigger.py +++ b/airflow/models/trigger.py @@ -244,8 +244,7 @@ def submit_event(cls, trigger_id, event, session: Session = NEW_SESSION) -> None trigger = session.scalars(select(cls).where(cls.id == trigger_id)).one() for asset in trigger.assets: AssetManager.register_asset_change( - asset=asset.to_public(), - session=session, + asset=asset.to_public(), session=session, extra={"from_trigger": True} ) @classmethod diff --git a/tests/models/test_trigger.py b/tests/models/test_trigger.py index 97c2b10208253..c9966b1cb0f24 100644 --- a/tests/models/test_trigger.py +++ b/tests/models/test_trigger.py @@ -185,6 +185,9 @@ def test_submit_event(session, create_task_instance): # Check that the asset has received an event assert session.query(AssetEvent).filter_by(asset_id=asset.id).count() == 1 + event = session.query(AssetEvent).filter_by(asset_id=asset.id).first() + assert event.extra == {"from_trigger": True} + def test_submit_failure(session, create_task_instance): """