Replace requests with httpx

airflow/providers/airbyte/CHANGELOG.rst

@@ -19,6 +19,17 @@
 Changelog
 ---------
 
+2.0.0
+.....
+
+Breaking changes
+~~~~~~~~~~~~~~~~
+
+The Airbyte Hook derives now from the new version of Hook in Http Provider.
+
+TODO: Add extra dependency to 2.0.0 version of http Provider.
+
+
 1.0.0
 .....
 

airflow/providers/airbyte/provider.yaml

@@ -22,6 +22,7 @@ description: |
   `Airbyte <https://airbyte.io/>`__
 
 versions:
+  - 2.0.0
   - 1.0.0
 
 integrations:

airflow/providers/apache/livy/CHANGELOG.rst

@@ -19,6 +19,14 @@
 Changelog
 ---------
 
+2.0.0
+.....
+
+The Livy Hook derives now from the new version of Http Hook in Http Provider.
+
+TODO: Add extra dependency to 2.0.0 version of http Provider.
+
+
 1.1.0
 .....
 

airflow/providers/apache/livy/hooks/livy.py

@@ -21,7 +21,7 @@
 from enum import Enum
 from typing import Any, Dict, List, Optional, Sequence, Union
 
-import requests
+import httpx
 
 from airflow.exceptions import AirflowException
 from airflow.providers.http.hooks.http import HttpHook
@@ -75,19 +75,25 @@ def __init__(
         super().__init__(http_conn_id=livy_conn_id)
         self.extra_options = extra_options or {}
 
-    def get_conn(self, headers: Optional[Dict[str, Any]] = None) -> Any:
+    def get_conn(
+        self, headers: Optional[Dict[Any, Any]] = None, verify: bool = True, proxies=None, cert=None
+    ) -> httpx.Client:
         """
         Returns http session for use with requests
 
         :param headers: additional headers to be passed through as a dictionary
         :type headers: dict
-        :return: requests session
-        :rtype: requests.Session
+        :param verify: whether to verify SSL during the connection (only use for testing)
+        :param proxies: A dictionary mapping proxy keys to proxy
+        :param cert: client An SSL certificate used by the requested host
+            to authenticate the client. Either a path to an SSL certificate file, or
+            two-tuple of (certificate file, key file), or a three-tuple of (certificate
+            file, key file, password).
         """
         tmp_headers = self._def_headers.copy()  # setting default headers
         if headers:
             tmp_headers.update(headers)
-        return super().get_conn(tmp_headers)
+        return super().get_conn(tmp_headers, verify, proxies, cert)
 
     def run_method(
         self,
@@ -108,7 +114,7 @@ def run_method(
         :param headers: headers
         :type headers: dict
         :return: http response
-        :rtype: requests.Response
+        :rtype: httpx.Response
         """
         if method not in ('GET', 'POST', 'PUT', 'DELETE', 'HEAD'):
             raise ValueError(f"Invalid http method '{method}'")
@@ -142,7 +148,7 @@ def post_batch(self, *args: Any, **kwargs: Any) -> Any:
 
         try:
             response.raise_for_status()
-        except requests.exceptions.HTTPError as err:
+        except httpx.HTTPStatusError as err:
             raise AirflowException(
                 "Could not submit batch. Status code: {}. Message: '{}'".format(
                     err.response.status_code, err.response.text
@@ -172,7 +178,7 @@ def get_batch(self, session_id: Union[int, str]) -> Any:
 
         try:
             response.raise_for_status()
-        except requests.exceptions.HTTPError as err:
+        except httpx.HTTPStatusError as err:
             self.log.warning("Got status code %d for session %d", err.response.status_code, session_id)
             raise AirflowException(
                 f"Unable to fetch batch with id: {session_id}. Message: {err.response.text}"
@@ -196,7 +202,7 @@ def get_batch_state(self, session_id: Union[int, str]) -> BatchState:
 
         try:
             response.raise_for_status()
-        except requests.exceptions.HTTPError as err:
+        except httpx.HTTPStatusError as err:
             self.log.warning("Got status code %d for session %d", err.response.status_code, session_id)
             raise AirflowException(
                 f"Unable to fetch batch with id: {session_id}. Message: {err.response.text}"
@@ -223,7 +229,7 @@ def delete_batch(self, session_id: Union[int, str]) -> Any:
 
         try:
             response.raise_for_status()
-        except requests.exceptions.HTTPError as err:
+        except httpx.HTTPStatusError as err:
             self.log.warning("Got status code %d for session %d", err.response.status_code, session_id)
             raise AirflowException(
                 "Could not kill the batch with session id: {}. Message: {}".format(

airflow/providers/apache/livy/provider.yaml

@@ -22,6 +22,7 @@ description: |
     `Apache Livy <https://livy.apache.org/>`__
 
 versions:
+  - 2.0.0
   - 1.1.0
   - 1.0.1
   - 1.0.0

airflow/providers/http/CHANGELOG.rst

@@ -19,6 +19,36 @@
 Changelog
 ---------
 
+
+2.0.0
+.....
+
+Breaking changes
+~~~~~~~~~~~~~~~~
+
+Due to licencing issues, the HTTP provider switched from ``requests`` to ``httpx`` library.
+In case you use an authentication method different than default Basic Authenticaton,
+you will need to change it to use the httpx-compatible one.
+
+HttpHook's run() method passes any kwargs passed to underlying httpx.Requests object rather than
+to requests.Requests. They largely compatible (for example json kwarg is supported in both) but not fully.
+The ``content`` and ``auth`` parameters are not supported in ``httpx``.
+
+The "verify", "proxie" and "cert" extra parameters
+
+The get_conn() method of HttpHook returns ``httpx.Client`` rather than ``requests.Session``.
+
+The get_conn() method of HttpHook has extra parameters: verify ,proxies and cert which are executed by the
+run() method, so if your hook derives from the HttpHook it should be updated.
+
+The ``run_and_check`` method is gone. PreparedRequests are not supported in ``httpx`` and this method
+used it. Instead, run method simply creates and executes the full request instance.
+
+While ``httpx`` does not support ``REQUESTS_CA_BUNDLE`` variable overriding ca crt, we backported
+the requests behaviour to HTTPHook and it respects the variable if set. More details on that variable
+can be found `here <https://docs.python-requests.org/en/master/user/advanced/#ssl-cert-verification`_
+
+
 1.1.1
 .....
 

airflow/providers/http/example_dags/example_http.py

@@ -75,7 +75,7 @@
     task_id='get_op_response_filter',
     method='GET',
     endpoint='get',
-    response_filter=lambda response: response.json()['nested']['property'],
+    response_filter=lambda response: response.json()['headers']['Host'],
     dag=dag,
 )
 # [END howto_operator_http_task_get_op_response_filter]