amoghrajesh commented on PR #67248:
URL: https://github.com/apache/airflow/pull/67248#issuecomment-4552829800
@jroachgolf84 a few things:
1. The elif branches added to `triggerer_job_runner.py` should stay. Without
them, any asset state call from a trigger crashes. That code is correct.
2. What changes is how trigger authors use it
Instead of:
`state = AssetState(name=self.asset_name)`
It will be:
`watermark = self.asset_states["orders"].get("watermark")`
So AssetState as a public class can be dropped.
3. What should the key be?
`self.asset_states["orders"]`, but "orders" is the asset name. For assets
declared with `AssetUriRef`, the name might be empty. Safest to use name when
available, and uri as fallback - same way as accessor.
And yes, `dict[str, AssetStateAccessor]` is the right shape as I see it. One
thing to figure out on your end: where does the triggerer get the asset list to
populate it? For event-driven triggers the asset-to-trigger association is
probably on the trigger model. For task-deferred triggers it should come from
`task_instance.task.outlets`. Either way the triggerer has this info at the
point it sets self.task_instance, just needs wiring
--
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]
