Add support for httpx as backend

.github/workflows/python-package.yml

@@ -53,6 +53,13 @@ jobs:
         COLOR: 'yes'
       run: |
         make mototest
+    - name: Run unittests without httpx
+      if: matrix.python-version == '3.11'
+      env:
+        COLOR: 'yes'
+      run: |
+        pip uninstall --yes httpx
+        HTTP_BACKEND='aiohttp' FLAGS='--cov-append' make mototest
     - name: Upload coverage to Codecov
       if: matrix.python-version == '3.11'
       uses: codecov/[email protected]

CONTRIBUTING.rst

@@ -22,6 +22,7 @@ For example, using *virtualenvwrapper* commands could look like::
 
 After that, please install libraries required for development::
 
+    $ pip install pip-tools
     $ pip-compile requirements-dev.in
     $ pip-sync requirements-dev.txt
 

Makefile

@@ -1,6 +1,8 @@
 # Some simple testing tasks (sorry, UNIX only).
 
-FLAGS=
+FLAGS?=
+# ?= is assignment conditional on it not being set
+HTTP_BACKEND?='all'
 
 pre-commit flake: checkrst
 	pre-commit run --all
@@ -22,7 +24,7 @@ cov cover coverage: pre-commit
 mototest:
 	docker pull alpine
 	docker pull lambci/lambda:python3.8
-	BOTO_CONFIG=/dev/null python -Wd -X tracemalloc=5 -X faulthandler -m pytest -vv -m moto -n auto --cov-report term --cov-report html --cov-report xml --cov=aiobotocore --cov=tests --log-cli-level=DEBUG aiobotocore tests
+	BOTO_CONFIG=/dev/null python -Wd -X tracemalloc=5 -X faulthandler -m pytest -vv -m moto -n auto --cov-report term --cov-report html --cov-report xml --cov=aiobotocore --cov=tests --log-cli-level=DEBUG --http-backend=$(HTTP_BACKEND) $(FLAGS) aiobotocore tests
 	@echo "open file://`pwd`/htmlcov/index.html"
 
 clean:

README.rst

@@ -167,10 +167,11 @@ secret accessible via environment variables:
 ::
 
     $ pip install pip-tools
-    $ pip-compile requirements-dev.txt
+    $ pip-compile requirements-dev.in
     $ pip-sync requirements-dev.txt
     $ export AWS_ACCESS_KEY_ID=xxx
     $ export AWS_SECRET_ACCESS_KEY=xxx
+    $ export AWS_DEFAULT_REGION=xxx # e.g. us-west-2
 
 Execute tests suite:
 

aiobotocore/_endpoint_helpers.py

@@ -4,6 +4,11 @@
 import botocore.retryhandler
 import wrapt
 
+try:
+    import httpx
+except ImportError:
+    httpx = None
+
 # Monkey patching: We need to insert the aiohttp exception equivalents
 # The only other way to do this would be to have another config file :(
 _aiohttp_retryable_exceptions = [
@@ -14,6 +19,15 @@
     asyncio.TimeoutError,
 ]
 
+if httpx is not None:
+    # TODO: Wild guesses after looking at https://pydoc.dev/httpx/latest/classIndex.html
+    # somebody with more network and/or httpx knowledge should revise this list.
+    _aiohttp_retryable_exceptions.extend(
+        (httpx.NetworkError, httpx.ConnectTimeout)
+    )
+
+
+# TODO [httpx]: determine retryable exceptions
 botocore.retryhandler.EXCEPTION_MAP['GENERAL_CONNECTION_ERROR'].extend(
     _aiohttp_retryable_exceptions
 )

aiobotocore/args.py

@@ -4,6 +4,8 @@
 import botocore.serialize
 from botocore.args import ClientArgsCreator
 
+from aiobotocore.httpsession import AIOHTTPSession
+
 from .config import AioConfig
 from .endpoint import AioEndpointCreator
 from .regions import AioEndpointRulesetResolver
@@ -67,8 +69,10 @@ def get_client_args(
         # aiobotocore addition
         if isinstance(client_config, AioConfig):
             connector_args = client_config.connector_args
+            http_session_cls = client_config.http_session_cls
         else:
             connector_args = None
+            http_session_cls = AIOHTTPSession
 
         new_config = AioConfig(connector_args, **config_kwargs)
         endpoint_creator = AioEndpointCreator(event_emitter)
@@ -79,9 +83,10 @@ def get_client_args(
             endpoint_url=endpoint_config['endpoint_url'],
             verify=verify,
             response_parser_factory=self._response_parser_factory,
+            timeout=(new_config.connect_timeout, new_config.read_timeout),
             max_pool_connections=new_config.max_pool_connections,
+            http_session_cls=http_session_cls,
             proxies=new_config.proxies,
-            timeout=(new_config.connect_timeout, new_config.read_timeout),
             socket_options=socket_options,
             client_cert=new_config.client_cert,
             proxies_config=new_config.proxies_config,

aiobotocore/awsrequest.py

@@ -28,3 +28,31 @@
     @property
     def text(self):
         return self._text_prop()
+
+
+class HttpxAWSResponse(AWSResponse):
+    # Unlike AWSResponse, these return awaitables
+
+    async def _content_prop(self):
+        """Content of the response as bytes."""
+
+        if self._content is None:
+            # NOTE: this will cache the data in self.raw
+            self._content = await self.raw.aread() or b''
+
+        return self._content
+
+    @property
+    def content(self):
+        return self._content_prop()
+
+    async def _text_prop(self):
+        encoding = botocore.utils.get_encoding_from_headers(self.headers)
+        if encoding:
+            return (await self.content).decode(encoding)
+        else:
+            return (await self.content).decode('utf-8')
+
+    @property
+    def text(self):
+        return self._text_prop()

aiobotocore/config.py

@@ -3,13 +3,18 @@
 import botocore.client
 from botocore.exceptions import ParamValidationError
 
+from aiobotocore.httpsession import AIOHTTPSession, HttpxSession
+
 
 class AioConfig(botocore.client.Config):
-    def __init__(self, connector_args=None, **kwargs):
+    def __init__(
+        self, connector_args=None, http_session_cls=AIOHTTPSession, **kwargs
+    ):
         super().__init__(**kwargs)
 
-        self._validate_connector_args(connector_args)
+        self._validate_connector_args(connector_args, http_session_cls)
         self.connector_args = copy.copy(connector_args)
+        self.http_session_cls = http_session_cls
         if not self.connector_args:
             self.connector_args = dict()
 
@@ -27,13 +32,17 @@ def merge(self, other_config):
         return AioConfig(self.connector_args, **config_options)
 
     @staticmethod
-    def _validate_connector_args(connector_args):
+    def _validate_connector_args(connector_args, http_session_cls):
         if connector_args is None:
             return
 
         for k, v in connector_args.items():
             # verify_ssl is handled by verify parameter to create_client
             if k == 'use_dns_cache':
+                if http_session_cls is HttpxSession:
+                    raise ParamValidationError(
+                        report='Httpx does not support dns caching. https://github.com/encode/httpx/discussions/2211'
+                    )
                 if not isinstance(v, bool):
                     raise ParamValidationError(
                         report=f'{k} value must be a boolean'
@@ -44,6 +53,10 @@ def _validate_connector_args(connector_args):
                         report=f'{k} value must be a float/int or None'
                     )
             elif k == 'force_close':
+                if http_session_cls is HttpxSession:
+                    raise ParamValidationError(
+                        report=f'Httpx backend does not currently support {k}.'
+                    )
                 if not isinstance(v, bool):
                     raise ParamValidationError(
                         report=f'{k} value must be a boolean'
@@ -59,6 +72,10 @@ def _validate_connector_args(connector_args):
             elif k == "resolver":
                 from aiohttp.abc import AbstractResolver
 
+                if http_session_cls is HttpxSession:
+                    raise ParamValidationError(
+                        report=f'Httpx backend does not support {k}.'
+                    )
                 if not isinstance(v, AbstractResolver):
                     raise ParamValidationError(
                         report=f'{k} must be an instance of a AbstractResolver'

aiobotocore/endpoint.py

@@ -1,4 +1,7 @@
+from __future__ import annotations
+
 import asyncio
+from typing import Any
 
 from botocore.endpoint import (
     DEFAULT_TIMEOUT,
@@ -13,14 +16,22 @@
     logger,
 )
 from botocore.hooks import first_non_none_response
-from urllib3.response import HTTPHeaderDict
+from requests.models import Response
+from urllib3._collections import HTTPHeaderDict
 
 from aiobotocore.httpchecksum import handle_checksum_body
 from aiobotocore.httpsession import AIOHTTPSession
 from aiobotocore.response import StreamingBody
 
+try:
+    import httpx
+except ImportError:
+    httpx = None
+
 
-async def convert_to_response_dict(http_response, operation_model):
+async def convert_to_response_dict(
+    http_response: Response, operation_model
+) -> dict[str, Any]:
     """Convert an HTTP response object to a request dict.
 
     This converts the requests library's HTTP response object to
@@ -36,15 +47,19 @@ async def convert_to_response_dict(http_response, operation_model):
         * body (string or file-like object)
 
     """
-    response_dict = {
+    if httpx and isinstance(http_response.raw, httpx.Response):
+        raw_headers = http_response.raw.headers.raw
+    else:  # aiohttp.ClientResponse
+        raw_headers = http_response.raw.raw_headers
+    response_dict: dict[str, Any] = {
         # botocore converts keys to str, so make sure that they are in
         # the expected case. See detailed discussion here:
         # https://github.com/aio-libs/aiobotocore/pull/116
         # aiohttp's CIMultiDict camel cases the headers :(
         'headers': HTTPHeaderDict(
             {
                 k.decode('utf-8').lower(): v.decode('utf-8')
-                for k, v in http_response.raw.raw_headers
+                for k, v in raw_headers
             }
         ),
         'status_code': http_response.status_code,
@@ -57,8 +72,11 @@ async def convert_to_response_dict(http_response, operation_model):
     elif operation_model.has_event_stream_output:
         response_dict['body'] = http_response.raw
     elif operation_model.has_streaming_output:
-        length = response_dict['headers'].get('content-length')
-        response_dict['body'] = StreamingBody(http_response.raw, length)
+        if httpx and isinstance(http_response.raw, httpx.Response):
+            response_dict['body'] = http_response.raw
+        else:
+            length = response_dict['headers'].get('content-length')
+            response_dict['body'] = StreamingBody(http_response.raw, length)
     else:
         response_dict['body'] = await http_response.content
     return response_dict
@@ -273,7 +291,7 @@ async def _needs_retry(
             return False
         else:
             # Request needs to be retried, and we need to sleep
-            # for the specified number of times.
+            # for the specified number of seconds.
             logger.debug(
                 "Response received to retry, sleeping for %s seconds",
                 handler_response,