Add asset partition sensor

airflow-core/newsfragments/67941.feature.rst

@@ -0,0 +1 @@
+Deferred triggers can now look up asset events through the triggerer, so a trigger may wait for an asset event (optionally filtered by ``partition_key``) instead of holding a worker slot.

airflow-core/src/airflow/jobs/triggerer_job_runner.py

@@ -73,6 +73,8 @@
     DeleteXCom,
     DRCount,
     ErrorResponse,
+    GetAssetEventByAsset,
+    GetAssetEventByAssetAlias,
     GetAssetStateStoreByName,
     GetAssetStateStoreByUri,
     GetConnection,
@@ -108,6 +110,8 @@
     handle_delete_asset_state_store_by_uri,
     handle_delete_variable,
     handle_delete_xcom,
+    handle_get_asset_event_by_asset,
+    handle_get_asset_event_by_asset_alias,
     handle_get_asset_state_store_by_name,
     handle_get_asset_state_store_by_uri,
     handle_get_connection,
@@ -396,6 +400,8 @@ def from_api_response(cls, response: HITLDetailResponse) -> HITLDetailResponseRe
     | SetXCom
     | GetTICount
     | GetTaskStates
+    | GetAssetEventByAsset
+    | GetAssetEventByAssetAlias
     | ClearAssetStateStoreByName
     | ClearAssetStateStoreByUri
     | DeleteAssetStateStoreByName
@@ -641,6 +647,10 @@ def _handle_request(self, msg: ToTriggerSupervisor, log: FilteringBoundLogger, r
             resp, dump_opts = handle_get_dr_count(self.client, msg)
         elif isinstance(msg, GetDagRunState):
             resp, dump_opts = handle_get_dag_run_state(self.client, msg)
+        elif isinstance(msg, GetAssetEventByAsset):
+            resp, dump_opts = handle_get_asset_event_by_asset(self.client, msg)
+        elif isinstance(msg, GetAssetEventByAssetAlias):
+            resp, dump_opts = handle_get_asset_event_by_asset_alias(self.client, msg)
 
         elif isinstance(msg, GetTICount):
             resp, dump_opts = handle_get_ti_count(self.client, msg)

airflow-core/tests/unit/jobs/test_triggerer_job.py

@@ -77,16 +77,19 @@
 from airflow.providers.standard.triggers.temporal import DateTimeTrigger, TimeDeltaTrigger
 from airflow.sdk import DAG, Asset, BaseHook, BaseOperator
 from airflow.sdk.api.client import Client
-from airflow.sdk.api.datamodels._generated import AssetStateStoreResponse
+from airflow.sdk.api.datamodels._generated import AssetEventsResponse, AssetStateStoreResponse
 from airflow.sdk.exceptions import ErrorType
 from airflow.sdk.execution_time import supervisor
 from airflow.sdk.execution_time.comms import (
+    AssetEventsResult,
     AssetStateStoreResult,
     ClearAssetStateStoreByName,
     ClearAssetStateStoreByUri,
     DeleteAssetStateStoreByName,
     DeleteAssetStateStoreByUri,
     ErrorResponse,
+    GetAssetEventByAsset,
+    GetAssetEventByAssetAlias,
     GetAssetStateStoreByName,
     GetAssetStateStoreByUri,
     OKResponse,
@@ -986,6 +989,73 @@ def test_clear_by_uri(self, supervisor):
         supervisor.send_msg.assert_called_once_with(OKResponse(ok=True), request_id=7, error=None)
 
 
+class TestTriggerSupervisorAssetEvents:
+    """Supervisor side of asset-event lookups made by deferred triggers.
+
+    A trigger that queries asset events reaches the supervisor twice: the frame is first
+    validated against the ``ToTriggerSupervisor`` union, then dispatched by
+    ``_handle_request``. Both are asserted here — a message missing from the union fails
+    to decode long before any handler branch runs.
+    """
+
+    @pytest.fixture
+    def supervisor(self, jobless_supervisor, mocker):
+        jobless_supervisor.client = mocker.MagicMock(spec=Client)
+        mocker.patch.object(TriggerRunnerSupervisor, "send_msg")
+        return jobless_supervisor
+
+    @pytest.mark.parametrize(
+        ("msg", "expected_kwargs"),
+        [
+            pytest.param(
+                GetAssetEventByAsset(
+                    name="orders",
+                    uri="s3://warehouse/orders",
+                    partition_key="2024-01-01",
+                    ascending=False,
+                    limit=1,
+                ),
+                {
+                    "uri": "s3://warehouse/orders",
+                    "name": "orders",
+                    "after": None,
+                    "before": None,
+                    "ascending": False,
+                    "limit": 1,
+                    "partition_key": "2024-01-01",
+                    "partition_key_regexp_pattern": None,
+                    "extra": None,
+                },
+                id="by_asset",
+            ),
+            pytest.param(
+                GetAssetEventByAssetAlias(alias_name="orders_alias", partition_key="2024-01-01"),
+                {
+                    "alias_name": "orders_alias",
+                    "after": None,
+                    "before": None,
+                    "ascending": True,
+                    "limit": None,
+                    "partition_key": "2024-01-01",
+                    "partition_key_regexp_pattern": None,
+                    "extra": None,
+                },
+                id="by_asset_alias",
+            ),
+        ],
+    )
+    def test_asset_event_lookup_is_decodable_and_replies(self, supervisor, msg, expected_kwargs):
+        assert TriggerRunnerSupervisor.decoder.validate_python(msg.model_dump()) == msg
+
+        supervisor.client.asset_events.get.return_value = AssetEventsResponse(asset_events=[])
+        supervisor._handle_request(msg, log=MagicMock(spec=FilteringBoundLogger), req_id=7)
+
+        supervisor.client.asset_events.get.assert_called_once_with(**expected_kwargs)
+        supervisor.send_msg.assert_called_once_with(
+            AssetEventsResult(asset_events=[]), request_id=7, error=None, exclude_unset=True
+        )
+
+
 def test_trigger_lifecycle(spy_agency: SpyAgency, session, testing_dag_bundle):
     """
     Checks that the triggerer will correctly see a new Trigger in the database

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,66 @@
+ .. 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/operator:AssetPartitionSensor:
+
+AssetPartitionSensor
+====================
+
+Use the :class:`~airflow.providers.standard.sensors.asset.AssetPartitionSensor` to wait for an asset
+event carrying a specific partition key.
+
+A Dag scheduled with :class:`~airflow.sdk.PartitionedAssetTimetable` is triggered *by* matching asset
+partitions. This sensor covers the opposite direction: a **time-scheduled** Dag that needs to block
+until one named partition of an upstream asset has been produced.
+
+Requires Airflow 3.4 or newer — partition-key filtering of asset events is not available on earlier
+versions.
+
+.. exampleinclude:: /../src/airflow/providers/standard/example_dags/example_asset_partition_sensor.py
+    :language: python
+    :dedent: 4
+    :start-after: [START howto_sensor_asset_partition]
+    :end-before: [END howto_sensor_asset_partition]
+
+Bounding the lookup with ``after``
+----------------------------------
+
+By default the sensor succeeds on *any* event that carries ``partition_key``, including one produced
+long before the current run. That is what you want when the key is unique per event — a timestamp
+such as ``2024-01-01T05``. It is not what you want when keys repeat across runs, for example a region
+code like ``us``: an event from last week would satisfy the wait immediately.
+
+Pass ``after`` to scope the lookup to the current interval. Both ``partition_key`` and ``after`` are
+templated, so they can be derived from the run:
+
+.. code-block:: python
+
+    AssetPartitionSensor(
+        task_id="wait_for_region",
+        asset=regional_sales,
+        partition_key="us",
+        after="{{ data_interval_start }}",
+    )
+
+Deferrable mode
+---------------
+
+Waiting for an upstream partition can take a long time, so prefer ``deferrable=True`` to release the
+worker slot while waiting. The sensor pokes once up front and only defers when the partition has not
+arrived yet. Deferral is also the default when ``[operators] default_deferrable`` is set.

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
@@ -106,6 +107,7 @@ sensors:
       - airflow.providers.standard.sensors.python
       - airflow.providers.standard.sensors.filesystem
       - airflow.providers.standard.sensors.external_task
+      - airflow.providers.standard.sensors.asset
 hooks:
   - integration-name: Standard
     python-modules:
@@ -120,6 +122,7 @@ triggers:
       - airflow.providers.standard.triggers.file
       - airflow.providers.standard.triggers.temporal
       - airflow.providers.standard.triggers.hitl
+      - airflow.providers.standard.triggers.asset
 
 extra-links:
   - airflow.providers.standard.operators.trigger_dagrun.TriggerDagRunLink

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

@@ -0,0 +1,48 @@
+#
+# 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.
+from __future__ import annotations
+
+from airflow.providers.standard.operators.empty import EmptyOperator
+from airflow.providers.standard.sensors.asset import AssetPartitionSensor
+from airflow.sdk import DAG, Asset, timezone
+
+hourly_sales = Asset(uri="s3://warehouse/sales/hourly", name="hourly_sales")
+
+with DAG(
+    dag_id="example_asset_partition_sensor",
+    start_date=timezone.datetime(2026, 1, 1),
+    schedule="@hourly",
+    catchup=False,
+    tags=["example", "asset", "partition"],
+) as dag:
+    # [START howto_sensor_asset_partition]
+    # ``after`` scopes the lookup to this run's interval. Without it, an event carrying the same
+    # partition key from an earlier run would satisfy the wait — harmless for keys that are unique
+    # per event (a timestamp), but wrong for keys that repeat (a region code).
+    wait_for_hourly_partition = AssetPartitionSensor(
+        task_id="wait_for_hourly_partition",
+        asset=hourly_sales,
+        partition_key="{{ data_interval_start.strftime('%Y-%m-%dT%H') }}",
+        after="{{ data_interval_start }}",
+        deferrable=True,
+    )
+    # [END howto_sensor_asset_partition]
+
+    summarize = EmptyOperator(task_id="summarize_hourly_sales")
+
+    wait_for_hourly_partition >> summarize

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

@@ -67,3 +67,7 @@ class HITLTimeoutError(HITLTriggerEventError):
 
 class HITLRejectException(AirflowException):
     """Raised when an ApprovalOperator receives a "Reject" response when fail_on_reject is set to True."""
+
+
+class AssetPartitionTriggerEventError(AirflowException):
+    """Raised when AssetPartitionTrigger reports an error event."""

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",
                 ],
             }
         ],
@@ -75,6 +76,7 @@ def get_provider_info():
                     "airflow.providers.standard.sensors.python",
                     "airflow.providers.standard.sensors.filesystem",
                     "airflow.providers.standard.sensors.external_task",
+                    "airflow.providers.standard.sensors.asset",
                 ],
             }
         ],
@@ -96,6 +98,7 @@ def get_provider_info():
                     "airflow.providers.standard.triggers.file",
                     "airflow.providers.standard.triggers.temporal",
                     "airflow.providers.standard.triggers.hitl",
+                    "airflow.providers.standard.triggers.asset",
                 ],
             }
         ],