Add AssetEventSensor to standard provider

devel-common/src/tests_common/test_utils/version_compat.py

@@ -42,6 +42,7 @@ def get_base_airflow_version_tuple() -> tuple[int, int, int]:
 AIRFLOW_V_3_2_PLUS = get_base_airflow_version_tuple() >= (3, 2, 0)
 AIRFLOW_V_3_2_2_PLUS = get_base_airflow_version_tuple() >= (3, 2, 2)
 AIRFLOW_V_3_3_PLUS = get_base_airflow_version_tuple() >= (3, 3, 0)
+AIRFLOW_V_3_4_PLUS = get_base_airflow_version_tuple() >= (3, 4, 0)
 
 if AIRFLOW_V_3_1_PLUS:
     from airflow.sdk import PokeReturnValue, timezone
@@ -64,6 +65,7 @@ def get_base_airflow_version_tuple() -> tuple[int, int, int]:
     "AIRFLOW_V_3_1_PLUS",
     "AIRFLOW_V_3_2_PLUS",
     "AIRFLOW_V_3_3_PLUS",
+    "AIRFLOW_V_3_4_PLUS",
     "NOTSET",
     "XCOM_RETURN_KEY",
     "ArgNotSet",

providers/standard/docs/sensors/asset.rst

@@ -0,0 +1,87 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+ ..   http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+
+
+.. _howto/sensor:AssetEventSensor:
+
+AssetEventSensor
+================
+
+Use the :class:`~airflow.providers.standard.sensors.asset.AssetEventSensor` to wait for
+asset events matching a set of filters to reach an expected count.
+
+.. note::
+
+    This sensor requires **Apache Airflow 3.4+**, because the ``partition_key``,
+    ``partition_key_regexp_pattern`` and ``extra`` asset-event filters it relies on are only
+    available in Airflow 3.4 and later.
+
+Basic usage
+-----------
+
+Point the sensor at an :class:`~airflow.sdk.Asset` (or :class:`~airflow.sdk.AssetAlias`, or the raw
+``name`` / ``uri`` / ``alias_name``). By default the sensor succeeds once **at least one** matching
+event exists.
+
+.. exampleinclude:: /../src/airflow/providers/standard/example_dags/example_asset_sensor.py
+    :language: python
+    :dedent: 4
+    :start-after: [START example_asset_event_sensor]
+    :end-before: [END example_asset_event_sensor]
+
+Filtering events and expected count
+-----------------------------------
+
+Events can be narrowed with ``after`` / ``before`` (time range), ``ascending`` / ``limit``
+(ordering and cap), ``partition_key`` / ``partition_key_regexp_pattern`` and ``extra`` (key/value
+pairs contained in the event ``extra`` field). Use ``expected_count`` to control how many
+(processed) events are required to succeed: ``-1`` (the default) means "at least one", ``0`` means
+"exactly zero", and any other positive value requires an exact match.
+
+.. exampleinclude:: /../src/airflow/providers/standard/example_dags/example_asset_sensor.py
+    :language: python
+    :dedent: 4
+    :start-after: [START example_asset_event_sensor_filtered]
+    :end-before: [END example_asset_event_sensor_filtered]
+
+Post-processing results
+-----------------------
+
+Pass a ``process_result`` callable (or a dotted import path to one) to transform, deduplicate or
+filter the fetched events **before** the count check. It receives the list of asset events and must
+return a list; the processed events are pushed to XCom.
+
+.. exampleinclude:: /../src/airflow/providers/standard/example_dags/example_asset_sensor.py
+    :language: python
+    :dedent: 0
+    :start-after: [START example_asset_event_process_result]
+    :end-before: [END example_asset_event_process_result]
+
+Deferrable mode
+---------------
+
+Set ``deferrable=True`` to run the sensor in :ref:`deferrable mode <deferring/writing>`. In this
+mode ``process_result`` runs in the triggerer, so it must be a **top-level importable function**
+(defined in an installed module, not a lambda, nested function, or bound method) and its return
+value must be JSON-serializable.
+
+.. exampleinclude:: /../src/airflow/providers/standard/example_dags/example_asset_sensor.py
+    :language: python
+    :dedent: 4
+    :start-after: [START example_asset_event_sensor_async]
+    :end-before: [END example_asset_event_sensor_async]

providers/standard/provider.yaml

@@ -81,6 +81,7 @@ integrations:
       - /docs/apache-airflow-providers-standard/sensors/datetime.rst
       - /docs/apache-airflow-providers-standard/sensors/file.rst
       - /docs/apache-airflow-providers-standard/sensors/external_task_sensor.rst
+      - /docs/apache-airflow-providers-standard/sensors/asset.rst
 
 operators:
   - integration-name: Standard
@@ -98,6 +99,7 @@ operators:
 sensors:
   - integration-name: Standard
     python-modules:
+      - airflow.providers.standard.sensors.asset
       - airflow.providers.standard.sensors.date_time
       - airflow.providers.standard.sensors.time_delta
       - airflow.providers.standard.sensors.time
@@ -116,6 +118,7 @@ hooks:
 triggers:
   - integration-name: Standard
     python-modules:
+      - airflow.providers.standard.triggers.asset
       - airflow.providers.standard.triggers.external_task
       - airflow.providers.standard.triggers.file
       - airflow.providers.standard.triggers.temporal

providers/standard/src/airflow/providers/standard/example_dags/example_asset_sensor.py

@@ -0,0 +1,78 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Example DAG demonstrating the usage of the ``AssetEventSensor``."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from airflow.providers.common.compat.sdk import AirflowOptionalProviderFeatureException
+from airflow.providers.standard.version_compat import AIRFLOW_V_3_4_PLUS
+
+if not AIRFLOW_V_3_4_PLUS:
+    raise AirflowOptionalProviderFeatureException("AssetEventSensor needs Airflow 3.4+.")
+
+import pendulum
+
+from airflow.providers.standard.sensors.asset import AssetEventSensor
+from airflow.sdk import DAG, Asset
+
+my_asset = Asset("s3://bucket/my-file.csv")
+
+
+# [START example_asset_event_process_result]
+def latest_event_only(events: list[Any]) -> list[Any]:
+    """Keep only the most recent event (must be a top-level importable function in deferrable mode)."""
+    return events[-1:]
+
+
+# [END example_asset_event_process_result]
+
+
+with DAG(
+    dag_id="example_asset_sensor",
+    schedule=None,
+    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
+    catchup=False,
+    tags=["example"],
+) as dag:
+    # [START example_asset_event_sensor]
+    wait_for_asset_event = AssetEventSensor(
+        task_id="wait_for_asset_event",
+        asset=my_asset,
+    )
+    # [END example_asset_event_sensor]
+
+    # [START example_asset_event_sensor_filtered]
+    wait_for_partition_events = AssetEventSensor(
+        task_id="wait_for_partition_events",
+        asset=my_asset,
+        partition_key="2021-01-01",
+        expected_count=3,
+    )
+    # [END example_asset_event_sensor_filtered]
+
+    # [START example_asset_event_sensor_async]
+    wait_for_asset_event_async = AssetEventSensor(
+        task_id="wait_for_asset_event_async",
+        asset=my_asset,
+        process_result=latest_event_only,
+        deferrable=True,
+    )
+    # [END example_asset_event_sensor_async]
+
+    [wait_for_asset_event, wait_for_partition_events, wait_for_asset_event_async]

providers/standard/src/airflow/providers/standard/exceptions.py

@@ -57,6 +57,14 @@ class DuplicateStateError(AirflowExternalTaskSensorException):
     """Raised when duplicate states are provided across allowed, skipped and failed states."""
 
 
+class AssetEventSensorException(AirflowException):
+    """Base exception for all AssetEventSensor related errors."""
+
+
+class UnexpectedAssetEventTriggerEventError(AssetEventSensorException):
+    """Raised when the AssetEventTrigger returns an unexpected event to the sensor."""
+
+
 class HITLTriggerEventError(Exception):
     """Raised when TriggerEvent contains error."""
 

providers/standard/src/airflow/providers/standard/get_provider_info.py

@@ -43,6 +43,7 @@ def get_provider_info():
                     "/docs/apache-airflow-providers-standard/sensors/datetime.rst",
                     "/docs/apache-airflow-providers-standard/sensors/file.rst",
                     "/docs/apache-airflow-providers-standard/sensors/external_task_sensor.rst",
+                    "/docs/apache-airflow-providers-standard/sensors/asset.rst",
                 ],
             }
         ],
@@ -67,6 +68,7 @@ def get_provider_info():
             {
                 "integration-name": "Standard",
                 "python-modules": [
+                    "airflow.providers.standard.sensors.asset",
                     "airflow.providers.standard.sensors.date_time",
                     "airflow.providers.standard.sensors.time_delta",
                     "airflow.providers.standard.sensors.time",
@@ -92,6 +94,7 @@ def get_provider_info():
             {
                 "integration-name": "Standard",
                 "python-modules": [
+                    "airflow.providers.standard.triggers.asset",
                     "airflow.providers.standard.triggers.external_task",
                     "airflow.providers.standard.triggers.file",
                     "airflow.providers.standard.triggers.temporal",