Add HttpExternalTaskSensor for Dags in remote Airflow deployments

providers/http/docs/index.rst

@@ -112,6 +112,25 @@ PIP package                                 Version required
 ``pydantic``                                ``>=2.11.0``
 ==========================================  ======================================
 
+Optional cross provider package dependencies
+--------------------------------------------
+
+Those are dependencies that might be needed in order to use all the features of the package.
+You need to install the specified provider distributions in order to use them.
+
+You can install such cross-provider dependencies when installing from PyPI. For example:
+
+.. code-block:: bash
+
+    pip install apache-airflow-providers-http[standard]
+
+
+========================================================================================================  ============
+Dependent package                                                                                         Extra
+========================================================================================================  ============
+`apache-airflow-providers-standard <https://airflow.apache.org/docs/apache-airflow-providers-standard>`_  ``standard``
+========================================================================================================  ============
+
 Optional dependencies
 ---------------------
 
@@ -123,11 +142,12 @@ Install them when installing from PyPI. For example:
     pip install apache-airflow-providers-http[srv]
 
 
-=======  ====================
-Extra    Dependencies
-=======  ====================
-``srv``  ``dnspython>=2.0.0``
-=======  ====================
+============  =============================================
+Extra         Dependencies
+============  =============================================
+``srv``       ``dnspython>=2.0.0``
+``standard``  ``apache-airflow-providers-standard>=1.19.0``
+============  =============================================
 
 Downloading official packages
 -----------------------------

providers/http/docs/operators.rst

@@ -44,6 +44,42 @@ This sensor can also be used in deferrable mode
     :start-after: [START howto_operator_http_http_sensor_check_deferrable]
     :end-before: [END howto_operator_http_http_sensor_check_deferrable]
 
+.. _howto/operator:HttpExternalTaskSensor:
+
+HttpExternalTaskSensor
+----------------------
+
+Use the :class:`~airflow.providers.http.sensors.external_task.HttpExternalTaskSensor` to wait for a Dag,
+task group, or task of a *remote* Airflow deployment. It accepts the same parameters and behaves the same
+as :class:`~airflow.providers.standard.sensors.external_task.ExternalTaskSensor`, but queries the REST API
+(v2) of the remote deployment instead of the local Airflow deployment. Both deployments need to run
+Airflow 3, and waiting for a task group requires Airflow 3.2 or later on the remote deployment.
+The sensor requires the ``standard`` extra (``apache-airflow-providers-http[standard]``).
+
+The ``http_conn_id`` connection points to the base URL of the remote Airflow API server, e.g.
+``https://airflow.example.com``. The connection is used to authenticate as follows:
+
+* With login and password, the credentials are exchanged for a JWT access token via the ``/auth/token``
+  endpoint of the remote deployment's auth manager.
+* With a password but no login, the password is sent as a bearer token as-is, e.g. an API token issued by a
+  managed Airflow service.
+* Without credentials, no authentication is added, but headers such as ``Authorization`` can be set via
+  the connection extra.
+
+The user on the remote deployment needs read access to the awaited Dag, its Dag runs and task instances.
+
+.. code-block:: python
+
+    from airflow.providers.http.sensors.external_task import HttpExternalTaskSensor
+
+    wait_for_remote_task = HttpExternalTaskSensor(
+        task_id="wait_for_remote_task",
+        http_conn_id="remote_airflow",
+        external_dag_id="remote_dag",
+        external_task_id="remote_task",
+        deferrable=True,
+    )
+
 .. _howto/operator:HttpOperator:
 
 HttpOperator

providers/http/provider.yaml

@@ -105,6 +105,7 @@ operators:
 sensors:
   - integration-name: Hypertext Transfer Protocol (HTTP)
     python-modules:
+      - airflow.providers.http.sensors.external_task
       - airflow.providers.http.sensors.http
 
 hooks:
@@ -118,6 +119,7 @@ notifications:
 triggers:
   - integration-name: Hypertext Transfer Protocol (HTTP)
     python-modules:
+      - airflow.providers.http.triggers.external_task
       - airflow.providers.http.triggers.http
 
 connection-types:

providers/http/pyproject.toml

@@ -77,13 +77,17 @@ dependencies = [
 "srv" = [
     "dnspython>=2.0.0",
 ]
+"standard" = [
+    "apache-airflow-providers-standard>=1.19.0",  # use next version
+]
 
 [dependency-groups]
 dev = [
     "apache-airflow",
     "apache-airflow-task-sdk",
     "apache-airflow-devel-common",
     "apache-airflow-providers-common-compat",
+    "apache-airflow-providers-standard",
     # Additional devel dependencies (do not remove this line and add extra development dependencies)
 ]
 

providers/http/src/airflow/providers/http/get_provider_info.py

@@ -44,7 +44,10 @@ def get_provider_info():
         "sensors": [
             {
                 "integration-name": "Hypertext Transfer Protocol (HTTP)",
-                "python-modules": ["airflow.providers.http.sensors.http"],
+                "python-modules": [
+                    "airflow.providers.http.sensors.external_task",
+                    "airflow.providers.http.sensors.http",
+                ],
             }
         ],
         "hooks": [
@@ -57,7 +60,10 @@ def get_provider_info():
         "triggers": [
             {
                 "integration-name": "Hypertext Transfer Protocol (HTTP)",
-                "python-modules": ["airflow.providers.http.triggers.http"],
+                "python-modules": [
+                    "airflow.providers.http.triggers.external_task",
+                    "airflow.providers.http.triggers.http",
+                ],
             }
         ],
         "connection-types": [

providers/http/src/airflow/providers/http/sensors/external_task.py

@@ -0,0 +1,102 @@
+#
+# 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 collections.abc import Sequence
+from functools import cached_property
+from typing import TYPE_CHECKING, Any
+
+from airflow.providers.http.hooks.http import HttpHook
+from airflow.providers.http.triggers.external_task import HttpExternalTaskTrigger, _AirflowApiClient
+from airflow.providers.standard.sensors.external_task import ExternalTaskSensor
+
+if TYPE_CHECKING:
+    import datetime
+
+    from airflow.providers.common.compat.sdk import Context
+
+
+class HttpExternalTaskSensor(ExternalTaskSensor):
+    """
+    Waits for a Dag, task group, or task of a remote Airflow deployment to complete for a logical date.
+
+    Behaves like :class:`~airflow.providers.standard.sensors.external_task.ExternalTaskSensor`,
+    but queries the REST API (v2) of the remote Airflow 3 deployment configured in the
+    :ref:`http connection<howto/connection:http>` instead of the local Airflow deployment.
+
+    Waiting for a task group requires the remote deployment to run Airflow 3.2 or later.
+
+    .. seealso::
+        For more information on how to use this sensor, take a look at the guide:
+        :ref:`howto/operator:HttpExternalTaskSensor`
+
+    :param http_conn_id: The :ref:`http connection<howto/connection:http>` of the remote
+        Airflow deployment. (templated)
+
+    All other parameters are the same as those of
+    :class:`~airflow.providers.standard.sensors.external_task.ExternalTaskSensor`.
+    """
+
+    template_fields = [*ExternalTaskSensor.template_fields, "http_conn_id"]
+    # ExternalDagLink would point to the local deployment.
+    operator_extra_links = []
+
+    def __init__(self, *, http_conn_id: str = HttpHook.default_conn_name, **kwargs: Any) -> None:
+        super().__init__(**kwargs)
+        self.http_conn_id = http_conn_id
+
+    @cached_property
+    def _client(self) -> _AirflowApiClient:
+        return _AirflowApiClient(self.http_conn_id)
+
+    def _get_dr_count(
+        self, context: Context, logical_dates: Sequence[datetime.datetime], states: list[str]
+    ) -> int:
+        return self._client.get_dr_count(self.external_dag_id, logical_dates, states)
+
+    def _get_ti_count(
+        self, context: Context, logical_dates: Sequence[datetime.datetime], states: list[str]
+    ) -> int:
+        if TYPE_CHECKING:
+            assert self.external_task_ids
+        return self._client.get_ti_count(self.external_dag_id, self.external_task_ids, logical_dates, states)
+
+    def _get_task_group_states(
+        self, context: Context, logical_dates: Sequence[datetime.datetime]
+    ) -> dict[str, dict[str, Any]]:
+        if TYPE_CHECKING:
+            assert self.external_task_group_id
+        return self._client.get_task_group_states(
+            self.external_dag_id, self.external_task_group_id, logical_dates
+        )
+
+    def _get_trigger(
+        self, context: Context, logical_dates: Sequence[datetime.datetime]
+    ) -> HttpExternalTaskTrigger:
+        return HttpExternalTaskTrigger(
+            http_conn_id=self.http_conn_id,
+            external_dag_id=self.external_dag_id,
+            external_task_group_id=self.external_task_group_id,
+            external_task_ids=self.external_task_ids,
+            allowed_states=self.allowed_states,
+            failed_states=self.failed_states,
+            skipped_states=self.skipped_states,
+            poke_interval=self.poke_interval,
+            soft_fail=self.soft_fail,
+            logical_dates=list(logical_dates),
+        )