seanghaeli commented on code in PR #75521:
URL: https://github.com/apache/airflow/pull/75521#discussion_r4383903038
##########
task-sdk/src/airflow/sdk/execution_time/context.py:
##########
@@ -1147,6 +1147,67 @@ def __repr__(self) -> str:
return f"<AssetStateStoreAccessors [{', '.join(parts)}]>"
+class TriggerAssetStateStoreAccessors(AssetStateStoreAccessors):
+ """
+ Asset state store accessors for a watcher trigger.
+
+ Watchers with the same trigger classpath and kwargs share one trigger, and
every event it
+ yields goes to all of its assets, so the trigger has one position to keep
no matter how many
+ assets it serves. Writes go to every asset; reads return the first asset
holding the key.
+ """
+
+ def _accessors(self) -> list[AssetStateStoreAccessor]:
+ # Sorted so reads pick the same asset on every triggerer, whatever
order the assets were loaded in.
+ return [self._by_name[n] for n in sorted(self._by_name)] + [
+ self._by_uri[u] for u in sorted(self._by_uri)
+ ]
+
+ def get(self, key: str, default: JsonValue = None) -> JsonValue:
+ """Return the value from the first asset holding ``key``, or
``default`` if none does."""
+ for accessor in self._accessors():
Review Comment:
Good catch. The copies stay in sync while the assets share the trigger, but
one that leaves and comes back can be behind. The asset state store table
already tracks "updated_at" so I'll return it from the execution API and take
the newest copy instead of the first.
--
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]
