aip-93-scoping: Adding AssetWatcher.asset functionality

[jroachgolf84]

Description

With AIP-103 adding an interface to persist state for Assets, AIP-93 shifts more towards implementing a way to use this state when building BaseEventTrigger for "Asset-watching". Currently, any Trigger defined for "Asset watching" is Asset-unaware. This PR adds the ability to make these Trigger Asset-aware.

The new pattern implemented here allows for an Asset object to be passed to the AssetWatcher (via the trigger). This makes much more sense; first, I define an Asset. Then, I define a way to "watch" that Asset.

from airflow.sdk import Asset, AssetWatcher
from plugins.triggers import GenericEventTrigger

...  # Other imports

# Defining the Asset first
random_integer_asset = Asset(name="random_integer_asset")

random_integer_watcher = AssetWatcher(
    name="random_integer_watcher",
    asset=random_integer_asset,  # Passing the Asset to the AssetWatcher
    trigger=GenericEventTrigger(
        target_number=7,
        waiter_delay=30
    )
)

with DAG(
    dag_id="random_integer_downstream_dag",
    start_date=datetime(2026, 1, 1),
    schedule=[random_integer_asset]
) as dag:
    ...

The majority of these changes are setting up for AIP-103, and the ability to use an asset_name and asset_uri to instantiate some object to retrieve and set state. The only changes that were required for the new paradigm to be enabled was the changes to task-sdk/src/airflow/sdk/definitions/asset/__init__.py, adding the asset field and adding the __attrs_post_init__ to AssetWatcher.

Future State

Eventually, I'd love to get to the following syntax, which is SIGNIFICANTLY more intuitive for DAG authors. This is just an example, and I'm sure the syntax would actually look a bit different.

from airflow.sdk import asset, DAG

...  # Other imports

@asset.watcher(
    name="random_integer_asset_watcher",
    waiter_delay=30
)
def random_integer_asset(target_number):
    generated_number: int = random.randint(0, 10)
    
    # Anything "truthy" yields a TriggerEvent and breaks execution 
    if generated_number == target_number:
        return {"generated_number": generated_number}


with DAG(
    dag_id="random_integer_downstream_dag",
    start_date=datetime(2026, 1, 1),
    schedule=[random_integer_asset(7)]
) as dag:
    ...

Testing

Note, no unit tests have been implemented yet; this PR is just providing a forum for discussion. Tests will be added after discussion.

No, an LLM was not used to generate this code/create this PR.

[jroachgolf84]

cc: @cmarteepants, @vikramkoka

[vincbeck]

I was wondering why you want to pass the asset to the asset watcher as opposed to passing the asset watcher to the asset like today? I do not see why it makes more sense (though I have not strong opinion on which way is better). However, this will also introduce some breaking changes for the user, we would need to deprecate the existing way to define asset watcher, update documentation etc. That is why I am asking, to me, if there is no strong argument on whether this way is better, I do not think we should do that.