diff --git a/providers/http/src/airflow/providers/http/operators/http.py b/providers/http/src/airflow/providers/http/operators/http.py index 1c5da7688e5e4..c232b754530c7 100644 --- a/providers/http/src/airflow/providers/http/operators/http.py +++ b/providers/http/src/airflow/providers/http/operators/http.py @@ -17,6 +17,7 @@ # under the License. from __future__ import annotations +import warnings from collections.abc import Callable, Sequence from typing import TYPE_CHECKING, Any @@ -24,7 +25,13 @@ from requests import Response from airflow.providers.common.compat.sdk import AirflowException, BaseHook, BaseOperator, conf -from airflow.providers.http.triggers.http import HttpResponseSerializer, HttpTrigger, serialize_auth_type +from airflow.providers.http.triggers.http import ( + NON_IDEMPOTENT_METHOD_WARNING_PATTERN, + HttpResponseSerializer, + HttpTrigger, + serialize_auth_type, + warn_if_method_not_idempotent, +) from airflow.utils.helpers import merge_dicts if TYPE_CHECKING: @@ -209,18 +216,36 @@ def paginate_sync(self, response: Response) -> Response | list[Response]: return all_responses def execute_async(self, context: Context) -> None: + trigger = self._build_http_trigger( + endpoint=self.endpoint, + headers=self.headers, + data=self.data, + extra_options=self.extra_options, + ) self.defer( - trigger=HttpTrigger( + trigger=trigger, + method_name="execute_complete", + ) + + def _build_http_trigger(self, **trigger_kwargs: Any) -> HttpTrigger: + warn_if_method_not_idempotent( + self.method, + subject="HttpOperator with deferrable=True", + execution_context="Deferrable mode executes the request in the Triggerer", + alternative="HttpSensor/EventSensor", + method_connector="and", + stacklevel=3, + ) + with warnings.catch_warnings(): + warnings.filterwarnings( + "ignore", category=UserWarning, message=NON_IDEMPOTENT_METHOD_WARNING_PATTERN + ) + return HttpTrigger( http_conn_id=self.http_conn_id, auth_type=serialize_auth_type(self._resolve_auth_type()), method=self.method, - endpoint=self.endpoint, - headers=self.headers, - data=self.data, - extra_options=self.extra_options, - ), - method_name="execute_complete", - ) + **trigger_kwargs, + ) def _resolve_auth_type(self) -> type[AuthBase] | type[BasicAuth] | None: """ @@ -300,13 +325,9 @@ def paginate_async( next_page_params = self.pagination_function(response) if not next_page_params: return self.process_response(context=context, response=all_responses) + trigger = self._build_http_trigger(**self._merge_next_page_parameters(next_page_params)) self.defer( - trigger=HttpTrigger( - http_conn_id=self.http_conn_id, - auth_type=serialize_auth_type(self._resolve_auth_type()), - method=self.method, - **self._merge_next_page_parameters(next_page_params), - ), + trigger=trigger, method_name="execute_complete", kwargs={"paginated_responses": all_responses}, ) diff --git a/providers/http/src/airflow/providers/http/triggers/http.py b/providers/http/src/airflow/providers/http/triggers/http.py index ec77d2634ee32..ce79452124bdc 100644 --- a/providers/http/src/airflow/providers/http/triggers/http.py +++ b/providers/http/src/airflow/providers/http/triggers/http.py @@ -21,6 +21,7 @@ import importlib import inspect import sys +import warnings from collections.abc import AsyncIterator from importlib import import_module from typing import TYPE_CHECKING, Any @@ -36,6 +37,9 @@ from airflow.providers.http.hooks.http import HttpAsyncHook from airflow.triggers.base import BaseTrigger, TriggerEvent +IDEMPOTENT_METHODS = {"GET", "HEAD", "OPTIONS", "PUT", "DELETE", "TRACE"} +NON_IDEMPOTENT_METHOD_WARNING_PATTERN = ".*may send duplicate requests.*" + if AIRFLOW_V_3_0_PLUS: from airflow.triggers.base import BaseEventTrigger else: @@ -60,6 +64,28 @@ def deserialize_auth_type(path: str | None) -> type | None: return getattr(import_module(module_path), cls_name) +def warn_if_method_not_idempotent( + method: str, + *, + subject: str, + execution_context: str, + alternative: str, + method_connector: str = "with", + stacklevel: int = 2, +) -> None: + if method.upper() in IDEMPOTENT_METHODS: + return + + warnings.warn( + f"{subject} {method_connector} method={method} may send duplicate requests if the Triggerer restarts. " + f"{execution_context}, which may be re-run on restart. " + f"Use only with idempotent methods or use {alternative} for polling. " + "See https://github.com/apache/airflow/issues/67945", + UserWarning, + stacklevel=stacklevel, + ) + + class HttpResponseSerializer: """Serializer for requests.Response objects used in deferred HTTP tasks.""" @@ -141,6 +167,13 @@ def __init__( self.headers = headers self.data = data self.extra_options = extra_options + warn_if_method_not_idempotent( + self.method, + subject="HttpTrigger", + execution_context="The trigger executes the request in the Triggerer", + alternative="HttpSensorTrigger/HttpEventTrigger", + stacklevel=2, + ) def serialize(self) -> tuple[str, dict[str, Any]]: """Serialize HttpTrigger arguments and classpath.""" diff --git a/providers/http/tests/unit/http/operators/test_http.py b/providers/http/tests/unit/http/operators/test_http.py index 50b66e5f61255..6dd82e8d06800 100644 --- a/providers/http/tests/unit/http/operators/test_http.py +++ b/providers/http/tests/unit/http/operators/test_http.py @@ -21,6 +21,7 @@ import contextlib import json import pickle +import warnings from types import SimpleNamespace from unittest import mock from unittest.mock import call, patch @@ -113,6 +114,25 @@ def test_async_defer_successfully(self, requests_mock): operator.execute({}) assert isinstance(exc.value.trigger, HttpTrigger), "Trigger is not a HttpTrigger" + def test_async_defer_warns_once_for_non_idempotent_method(self, monkeypatch): + captured = self._capture_defer(monkeypatch) + + with pytest.warns(UserWarning, match="HttpOperator with deferrable=True and method=POST") as warns: + HttpOperator(task_id="test_HTTP_op", deferrable=True).execute(context={}) + + assert len(warns) == 1 + assert "HttpTrigger" not in str(warns[0].message) + assert isinstance(captured["trigger"], HttpTrigger) + + def test_async_defer_does_not_warn_for_idempotent_method(self, monkeypatch): + captured = self._capture_defer(monkeypatch) + + with warnings.catch_warnings(): + warnings.simplefilter("error", UserWarning) + HttpOperator(task_id="test_HTTP_op", method="GET", deferrable=True).execute(context={}) + + assert isinstance(captured["trigger"], HttpTrigger) + def test_async_execute_successfully(self, requests_mock): operator = HttpOperator( task_id="test_HTTP_op", diff --git a/providers/http/tests/unit/http/triggers/test_http.py b/providers/http/tests/unit/http/triggers/test_http.py index e3338674af4c0..e0fd8d6d1dda4 100644 --- a/providers/http/tests/unit/http/triggers/test_http.py +++ b/providers/http/tests/unit/http/triggers/test_http.py @@ -17,6 +17,7 @@ # under the License. from __future__ import annotations +import warnings from asyncio import Future from http.cookies import SimpleCookie from typing import Any @@ -135,6 +136,17 @@ def test_serialization(self, trigger): "extra_options": TEST_EXTRA_OPTIONS, } + def test_warns_for_non_idempotent_method(self): + with pytest.warns(UserWarning, match="HttpTrigger with method=POST") as warns: + HttpTrigger(method="POST") + + assert len(warns) == 1 + + def test_does_not_warn_for_idempotent_method(self): + with warnings.catch_warnings(): + warnings.simplefilter("error", UserWarning) + HttpTrigger(method="GET") + @pytest.mark.asyncio @mock.patch(HTTP_PATH.format("HttpAsyncHook")) async def test_trigger_on_success_yield_successfully(self, mock_hook, trigger, client_response):