hussein-awala opened a new pull request, #70225: URL: https://github.com/apache/airflow/pull/70225
## 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` (#64610) - `extra` key/value filtering - time range / ordering / limit (`after`, `before`, `ascending`, `limit`) from #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`: succeed when the (processed) number of events equals it; `-1` (default) means **at least one**. - `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. ## Tests Unit tests for the sensor (poke exact/at-least-one/no-match, `process_result` affecting count, defer/short-circuit, `execute_complete`) and the trigger (serialize round-trip, `run()` fires on count match, applies imported `process_result`, sleeps otherwise). `ruff` + `mypy` clean. --- > [!NOTE] > Draft: docs (howto guide + operators-and-hooks reference entry) still to be added. -- 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]
