sjyangkevin commented on PR #50182: URL: https://github.com/apache/airflow/pull/50182#issuecomment-2861250325
> > Found a unit test case `airflow-core/tests/unit/models/test_taskinstance.py`, from my understanding the extra information should be available from the `inlet_events`, since the test passed. The test is implemented in task flow api, but seems my code is in traditional syntax. Will need further investigate into it. > > for using traditional syntax, you can access the value through `context` > > https://airflow.apache.org/docs/apache-airflow/stable/release_notes.html#improved-inlet-events-outlet-events-and-triggering-asset-events Hi Wei, thank you. I modified the DAG code to access the asset events from context based on the document you provided. However, the asset event associated with the alias is not in the `inlets`. I also tried to use TaskFlow API, but the get into the same situation. I've attached the DAG code in the end.  Also, I could try to explain it from the source code. From line 1606 to 1634 in `airflow-core/src/airflow/jobs/scheduler_job_runner.py`. ```python # here, we only query on events join with `DagScheduleAssetReference`, so I think when the DAG is scheduled on asset # alias, this query return empty results, and hence not populate the `consumed_asset_events` asset_events = session.scalars( select(AssetEvent) .join( DagScheduleAssetReference, AssetEvent.asset_id == DagScheduleAssetReference.asset_id, ) .where( DagScheduleAssetReference.dag_id == dag.dag_id, AssetEvent.timestamp <= triggered_date, AssetEvent.timestamp > func.coalesce(cte.c.previous_dag_run_run_after, date.min), ) ).all() dag_run = dag.create_dagrun( run_id=DagRun.generate_run_id( run_type=DagRunType.ASSET_TRIGGERED, logical_date=None, run_after=triggered_date ), logical_date=None, data_interval=None, run_after=triggered_date, run_type=DagRunType.ASSET_TRIGGERED, triggered_by=DagRunTriggeredByType.ASSET, state=DagRunState.QUEUED, creating_job_id=self.job.id, session=session, ) Stats.incr("asset.triggered_dagruns") dag_run.consumed_asset_events.extend(asset_events) session.execute(delete(AssetDagRunQueue).where(AssetDagRunQueue.target_dag_id == dag_run.dag_id)) ``` In `taskinstance.py`, the inlets seems only included the asset information if it is triggered by asset instead of asset alias.  Consumer DAG ```python from airflow.sdk import Asset, AssetAlias, DAG from airflow.providers.standard.operators.python import PythonOperator file_asset_alias = AssetAlias("file_asset_event") with DAG( dag_id="asset_alias_consumer", schedule=[file_asset_alias] ): def my_function(**context): print(context) inlet_events = context["inlet_events"] outlet_events = context["outlet_events"] print("Access triggering_asset_events in the consumer task") print("triggering_asset_events:", context["triggering_asset_events"]) print("Access inlet_events in the consumer task, from asset alias 'file_asset_event'") print("inlet_events:", inlet_events[AssetAlias(name="file_asset_event")]) print("inlet_events:", inlet_events.for_asset_alias(name="file_asset_event")) print("Access outlet_events in the consumer task, from asset alias 'file_asset_event'") print("outlet_events:", outlet_events[AssetAlias(name="file_asset_event")]) print("outlet_events:", outlet_events.for_asset_alias(name="file_asset_event")) my_task = PythonOperator( task_id="consumer_task", python_callable=my_function, inlets=[file_asset_alias] ) ``` -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
