AIP-103: Passing AssetStateAccessors through to BaseEventTrigger

[jroachgolf84]

Description

To ensure that classes inheriting from the BaseEventTrigger (typically used for Asset watching) are "Asset-aware", the work done in AIP-103 has been wired into the BaseEventTrigger class.

#65103 drafted an approach that "flipped" the definition of the Asset and AssetWatcher. However, for good reason, the approach was challenged. After conversation, the goal became to "pass" the AssetStateAccessors through to the BaseEventTrigger with some runtime magic.

With this model, defining an Asset and AssetWatcher remains the same as before.

from airflow.sdk import Asset, AssetWatcher, DAG, task
from datetime import datetime
from triggers.event_triggers import GenericEventTrigger


generic_asset_watcher = AssetWatcher(
    name="generic_asset_watcher",
    trigger=GenericEventTrigger(
        random_number=1,
        waiter_delay=15
    )
)

generic_asset = Asset(
    name="generic_asset",
    watchers=[generic_asset_watcher],
)


with DAG(
    dag_id="aip_93_scoping",
    start_date=datetime(2026, 1, 1),
    schedule=[generic_asset]
) as dag:

    @task
    def downstream_task():
        pass

    downstream_task()

The two PR's below were closed in favor of this PR:

• aip-93-scoping: Adding AssetWatcher.asset functionality #65103
• AIP-103: Passing Asset name/uri and AssetStateAccessor through to BaseEventTrigger #66595

Testing

No unit tests have been written for this logic yet. However, testing has been performed E2E locally with breeze. The trigger authored below is what was used for testing. When this ran, the Asset name and uri were output in the Triggerer logs.

class GenericEventTrigger(BaseEventTrigger):
    def __init__(
        self,
        random_number,
        waiter_delay,
        asset_name,
        **kwargs
    ):
        super().__init__(**kwargs)

        self.random_number = random_number
        self.waiter_delay = waiter_delay
        self.asset_name = asset_name

    def serialize(self) -> tuple[str, dict[str, Any]]:
        """Serialize the Trigger, including the func, params, and waiter_delay."""
        return (
            self.__class__.__module__ + "." + self.__class__.__qualname__,
            {
                "random_number": self.random_number,
                "waiter_delay": self.waiter_delay,
                "asset_name": self.asset_name,
            },
        )

    async def run(self) -> AsyncIterator[TriggerEvent]:
        """Logic that fires a TriggerEvent."""
        logging.info(f"***** asset_states: {self.asset_states}")

        while True:
            result = random.randint(0, 5)

            if result == self.random_number:
                logging.info("yield'ing TriggerEvent")
                yield TriggerEvent({"status": "success", "result": result})
                break

            logging.info(f"Sleeping for {self.waiter_delay} seconds")
            await asyncio.sleep(self.waiter_delay)

[jroachgolf84]

cc: @cmarteepants

[jroachgolf84]

cc: @vincbeck and @amoghrajesh, I closed the previous PR (#66595) after renaming the branch. We'll use this one instead.

[vincbeck]

One nit but overall I like it

[amoghrajesh]

The _handle_request should be handling this in here?

[jroachgolf84]

What do you mean?