Re: [PR] Make HttpSensor defer when response_check is provided [airflow]

["jroachgolf84 (via GitHub)" <gi...@apache.org>]

jroachgolf84 commented on code in PR #71922:
URL: https://github.com/apache/airflow/pull/71922#discussion_r4006386877


##########
providers/http/src/airflow/providers/http/sensors/http.py:
##########
@@ -158,22 +165,48 @@ def poke(self, context: Context) -> bool | PokeReturnValue:
         return True
 
     def execute(self, context: Context) -> Any:
-        if not self.deferrable or self.response_check:
+        if not self.deferrable:
             return super().execute(context=context)
-        if not self.poke(context):
-            self.defer(
-                timeout=timedelta(seconds=self.timeout),
-                trigger=HttpSensorTrigger(
-                    endpoint=self.endpoint,
-                    http_conn_id=self.http_conn_id,
-                    data=self.request_params,
-                    headers=self.headers,
-                    method=self.method,
-                    extra_options=self.extra_options,
-                    poke_interval=self.poke_interval,
-                ),
-                method_name="execute_complete",
-            )
+        result = self.poke(context)

Review Comment:
   ```suggestion
           result = self.poke(context)
           
   ```



##########
providers/http/src/airflow/providers/http/sensors/http.py:
##########
@@ -158,22 +165,48 @@ def poke(self, context: Context) -> bool | PokeReturnValue:
         return True
 
     def execute(self, context: Context) -> Any:
-        if not self.deferrable or self.response_check:

Review Comment:
   Seems like the `self.response_check` behvaior is being dropped? Is that an issue?



##########
providers/http/src/airflow/providers/http/triggers/http.py:
##########
@@ -220,6 +220,9 @@ class HttpSensorTrigger(BaseTrigger):
     :param extra_options: Additional kwargs to pass when creating a request.
         For example, ``run(json=obj)`` is passed as ``aiohttp.ClientSession().get(json=obj)``
     :param poke_interval: Time to sleep using asyncio
+    :param initial_delay: Time to sleep before the first request. Used when the

Review Comment:
   Is this a pattern that is being used elsewhere (`initial_delay`, that is)?



##########
providers/http/src/airflow/providers/http/sensors/http.py:
##########
@@ -158,22 +165,48 @@ def poke(self, context: Context) -> bool | PokeReturnValue:
         return True
 
     def execute(self, context: Context) -> Any:
-        if not self.deferrable or self.response_check:
+        if not self.deferrable:
             return super().execute(context=context)
-        if not self.poke(context):
-            self.defer(
-                timeout=timedelta(seconds=self.timeout),
-                trigger=HttpSensorTrigger(
-                    endpoint=self.endpoint,
-                    http_conn_id=self.http_conn_id,
-                    data=self.request_params,
-                    headers=self.headers,
-                    method=self.method,
-                    extra_options=self.extra_options,
-                    poke_interval=self.poke_interval,
-                ),
-                method_name="execute_complete",
-            )
+        result = self.poke(context)
+        if not result:
+            self._defer()
+        # Keep sync mode's contract of returning the xcom value from a truthy PokeReturnValue.
+        if isinstance(result, PokeReturnValue):
+            return result.xcom_value
+
+    def _defer(self, initial_delay: float = 0.0) -> None:
+        self.defer(
+            timeout=timedelta(seconds=self.timeout),
+            trigger=HttpSensorTrigger(
+                endpoint=self.endpoint,
+                http_conn_id=self.http_conn_id,
+                data=self.request_params,
+                headers=self.headers,
+                method=self.method,
+                extra_options=self.extra_options,
+                poke_interval=self.poke_interval,
+                initial_delay=initial_delay,
+            ),
+            method_name="execute_complete",
+        )
 
-    def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> None:
+    def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> Any:
+        if self.response_check:
+            from airflow.utils.operator_helpers import determine_kwargs
+
+            if not isinstance(event, dict) or "response" not in event:
+                raise ValueError(
+                    "The trigger event does not contain the HTTP response required to "
+                    "evaluate response_check. The deferred task was most likely resumed by a "
+                    "trigger serialized with an older version of the http provider."
+                )
+            response = HttpResponseSerializer.deserialize(event["response"])
+            kwargs = determine_kwargs(self.response_check, [response], context)
+            result = self.response_check(response, **kwargs)

Review Comment:
   ```suggestion
               result = self.response_check(response, **kwargs)
               
   ```



-- 
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: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org