fix(http): request_kwargs not passed from caller

HttpHooK.run() accepts request_kwargs, however HttpOperator and HttpSensor do not pass request_kwargs to HttpHooK.run()
apache · Oct 28, 2024 · 8a9d5d6 · 8a9d5d6
1 parent f6ce1fd
commit 8a9d5d6
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 3 deletions.
diff --git a/providers/src/airflow/providers/http/operators/http.py b/providers/src/airflow/providers/http/operators/http.py
@@ -143,6 +143,7 @@ def __init__(
         self.tcp_keep_alive_interval = tcp_keep_alive_interval
         self.deferrable = deferrable
         self.retry_args = retry_args
+        self.kwargs = kwargs
 
     @property
     def hook(self) -> HttpHook:
@@ -173,10 +174,21 @@ def execute_sync(self, context: Context) -> Any:
         self.log.info("Calling HTTP method")
         if self.retry_args:
             response = self.hook.run_with_advanced_retry(
-                self.retry_args, self.endpoint, self.data, self.headers, self.extra_options
+                self.retry_args,
+                self.endpoint,
+                self.data,
+                self.headers,
+                self.extra_options,
+                **self.kwargs,
             )
         else:
-            response = self.hook.run(self.endpoint, self.data, self.headers, self.extra_options)
+            response = self.hook.run(
+                self.endpoint,
+                self.data,
+                self.headers,
+                self.extra_options,
+                **self.kwargs,
+            )
         response = self.paginate_sync(response=response)
         return self.process_response(context=context, response=response)
 
@@ -303,7 +315,10 @@ def _merge_next_page_parameters(self, next_page_params: dict) -> dict:
             endpoint=next_page_params.get("endpoint") or self.endpoint,
             data=data,
             headers=merge_dicts(self.headers, next_page_params.get("headers", {})),
-            extra_options=merge_dicts(self.extra_options, next_page_params.get("extra_options", {})),
+            extra_options=merge_dicts(
+                self.extra_options, next_page_params.get("extra_options", {})
+            ),
+            **self.kwargs,
         )
 
 

diff --git a/providers/src/airflow/providers/http/sensors/http.py b/providers/src/airflow/providers/http/sensors/http.py
@@ -121,6 +121,7 @@ def __init__(
         self.tcp_keep_alive_count = tcp_keep_alive_count
         self.tcp_keep_alive_interval = tcp_keep_alive_interval
         self.deferrable = deferrable
+        self.kwargs = kwargs
 
     def poke(self, context: Context) -> bool:
         from airflow.utils.operator_helpers import determine_kwargs
@@ -141,6 +142,7 @@ def poke(self, context: Context) -> bool:
                 data=self.request_params,
                 headers=self.headers,
                 extra_options=self.extra_options,
+                **self.kwargs,
             )
 
             if self.response_check: