Add AssetEventSensor to standard provider

[hussein-awala]

Summary

Adds a new AssetEventSensor (and backing AssetEventTrigger) to the standard provider that waits for asset events matching a set of filters to reach an expected count.

It builds on the asset-event query filters recently merged into the API/Task SDK:

• partition_key / partition_key_regexp_pattern (Add partition_key and partition_key_regexp_pattern filters for asset events #64610)
• extra key/value filtering
• time range / ordering / limit (after, before, ascending, limit) from feat: Add lazy filtering support for inlet events by time range, ordering, and limit #54891

Because those filters are only available from Airflow 3.4, the sensor is gated to Airflow 3.4+.

Features

• Target an Asset / AssetAlias (or raw name / uri / alias_name).
• Filters: after, before, ascending, limit, partition_key, partition_key_regexp_pattern, extra.
• expected_count: how many (processed) events are required to succeed. -1 (default) means at least one (count >= 1); 0 means exactly zero; any other positive value requires an exact match (count == expected_count). Values < -1 are rejected. Note: if limit is set below an exact expected_count, the condition can never be satisfied and the sensor waits until it times out.
• process_result: an optional callable (or dotted import path) to transform / deduplicate / filter the fetched events before the count check. The processed events are pushed to XCom.
• Deferrable mode.

Deferrable design

process_result runs in the triggerer (Kafka AwaitMessageSensor style):

• In deferrable mode the callable is passed to the trigger as a dotted import-path string, import_string()'d inside the trigger, and applied there. It must therefore be a top-level importable function, and its output must be JSON-serializable so it can flow back via the TriggerEvent.
• Since it may be a plain (non-async) function, it is executed in a separate thread via sync_to_async(process_result, thread_sensitive=False).
• The count check gates firing: the trigger only yields success once the processed count satisfies expected_count, returning the processed events.

Event fetching reuses the Task SDK InletEventsAccessor; blocking calls in the triggerer are wrapped with asgiref.sync.sync_to_async (same approach as triggers/external_task.py).

Notes

• Uses airflow.sdk.execution_time.context.InletEventsAccessor, which is not (yet) a formally public API for arbitrary assets. It mirrors how external_task.py already imports from airflow.sdk.execution_time.*.
• Running a Dag-author-supplied callable in the triggerer follows the Kafka precedent; the import path must not come from a Connection/attacker-controlled source.

Docs

• Adds a howto guide (sensors/asset.rst) covering basic usage, filtering / expected_count, process_result, and deferrable mode, wired into the provider's how-to-guide reference.
• Adds an example_asset_sensor example DAG (version-gated to 3.4+) referenced by the guide.

Tests

• Sensor and trigger unit tests are backed by a DB harness (a SUPERVISOR_COMMS stand-in in conftest.py that runs the real execution-API SQL/filter paths against the test DB), so poke / run() are exercised against real asset events rather than mocks.
• Covers poke semantics (exact / at-least-one / exactly-zero / no-match), all filters incl. partition_key(_regexp)/extra/time-range/ordering/limit, process_result (callable + import-path, dedup/filter/expand/empty/raise), deferrable short-circuit vs. defer, execute_complete, argument validation, and serialization round-trips. ruff + mypy clean.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Cursor (Claude Opus 4.8) following the guidelines

[Copilot]

Pull request overview

Adds a new AssetEventSensor and AssetEventTrigger to the Standard provider to wait for asset events matching filter criteria until an expected count is reached (including a deferrable mode that runs optional process_result processing in the triggerer). This integrates with the Task SDK’s InletEventsAccessor-based asset-event querying and is version-gated to Airflow 3.4+.

Changes:

• Introduce AssetEventSensor + AssetEventTrigger with support for filters (after/before, ordering/limit, partition_key, partition_key_regexp_pattern, extra) and expected_count semantics.
• Add provider/version compatibility flags (AIRFLOW_V_3_4_PLUS) and wire new modules into provider metadata.
• Add unit tests (including DB-backed harness via a SUPERVISOR_COMMS stand-in) for sensor/trigger behavior and deferrable flow.

Reviewed changes

Copilot reviewed 10 out of 10 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
providers/standard/src/airflow/providers/standard/sensors/asset.py	New AssetEventSensor implementation (sync + deferrable), filter forwarding, process_result handling, and execute_complete.
providers/standard/src/airflow/providers/standard/triggers/asset.py	New AssetEventTrigger implementation + shared helpers for fetch/count/serialization.
providers/standard/src/airflow/providers/standard/exceptions.py	Add sensor-specific exception types for trigger event validation.
providers/standard/src/airflow/providers/standard/version_compat.py	Add AIRFLOW_V_3_4_PLUS feature gate for the provider.
providers/standard/src/airflow/providers/standard/get_provider_info.py	Register new sensor/trigger modules in provider info.
providers/standard/provider.yaml	Register new sensor/trigger modules in provider YAML metadata.
providers/standard/tests/unit/standard/conftest.py	Add DB-backed fixtures and SUPERVISOR_COMMS stand-in to exercise real SQL/filtering paths in unit tests.
providers/standard/tests/unit/standard/sensors/test_asset.py	Add sensor tests (poke semantics, process_result, deferrable short-circuit/defer behavior, filter forwarding).
providers/standard/tests/unit/standard/triggers/test_asset.py	Add trigger tests (serialization, polling loop, process_result import/behavior, DB-backed run).
devel-common/src/tests_common/test_utils/version_compat.py	Add AIRFLOW_V_3_4_PLUS for tests to skip appropriately.

[Lee-W]

Thanks! I'll take a deeper look, but I'd first like to understand why we need a sensor like this.

I thought we could always use a custom PartitionMapper to achieve something similar. Are there any use cases I'm missing? Thanks!