obarisk commented on PR #50182:
URL: https://github.com/apache/airflow/pull/50182#issuecomment-3272735105
@Lee-W
On airflow 3.0.6, there's no `triggering_asset_events` for a task in a
AssetAlias trigger dag.
Can we backport this to airflow 3.0+? 🙏🏼
Here's the code snippet.
```python
from pprint import pprint
from typing import Any
from airflow.sdk import Asset, AssetAlias, Context, dag, task
from airflow.sdk.execution_time.context import (
OutletEventAccessors,
TriggeringAssetEventsAccessor,
)
@dag(schedule=None, tags=["demo"])
def assetalias_producer():
@task(outlets=[AssetAlias("asset_alias_0")])
def producer(**context: Context):
outlets: list[Any] = context["outlets"]
outlet_events: OutletEventAccessors = context["outlet_events"]
run_after = context["dag_run"].run_after
for a in outlets:
pprint(a)
match a:
case AssetAlias():
if int(run_after.timestamp()) % 2 == 0:
outlet_events[a].add(Asset("local0", extra={"run_after":
run_after}))
else:
outlet_events[a].add(Asset("local1", extra={"n": 1}))
outlet_events[a].add(Asset("local2", extra={"run_after":
run_after}))
case Asset():
outlet_events[a].extra = {"rc": 100}
case _:
raise ValueError(f"unexpected outlet type: {type(a)}")
return 0
producer()
@dag(schedule=[AssetAlias("asset_alias_0")], tags=["demo"])
def assetalias_consumer():
@task
def consumer(**context: Context):
triggering_events: TriggeringAssetEventsAccessor = context[
"triggering_asset_events"
]
pprint(triggering_events[AssetAlias("asset_alias_0")])
consumer()
assetalias_producer()
assetalias_consumer()
```
--
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-unsubscr...@airflow.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
