A consumer DAG triggered by a data update may need the producer’s logical date to select the correct input file. Time-based values such as logical_date are not automatically available when the consumer is scheduled by an asset. section 4.4
triggering_asset_events is an Airflow context dictionary containing the events that triggered a consumer DAG. Its keys identify triggering assets, and each value is a list because an asset can produce multiple events. section 4.4
Use the event collection to recover producer-run information and pass it into downstream task parameters:
- Select an asset with values() | first, then select its event with | first. section 4.4
- Read source_dag_run.logical_date to identify the logical date associated with the producer DAG run. section 4.4
Deriving consumer paths from the triggering event
calculate_stats = PythonOperator( task_id="calculate_stats", python_callable=_calculate_stats, op_kwargs={ "input_path": "/data/events/{{ (triggering_asset_events.values() | first | first).source_dag_run.logical_date }}.json", "output_path": "/data/stats/{{ (triggering_asset_events.values() | first | first).source_dag_run.logical_date }}.csv", }, )
The shown source_dag_run approach is reported as broken in Airflow 3 because the attribute appears to have been removed during the Airflow 2-to-3 transition. A metadata-based workaround is indicated. section 4.4
