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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
