Add support for httpx as backend (#1085)

Author: jakkdl

.github/workflows/ci-cd.yml

@@ -41,14 +41,24 @@ jobs:
         os:
         - ubuntu-24.04
         - ubuntu-24.04-arm
+        no-httpx:
+        - false
         experimental:
         - false
         include:
+        - python-version: 3.13  # add a no-httpx run
+          os: ubuntu-24.04
+          no-httpx: true
+          experimental: false
+        # add experimental 3.14 runs. Move this to being a 3.14 python-version
+        # entry when stable.
         - python-version: 3.14
           os: ubuntu-24.04
+          no-httpx: false
           experimental: true
         - python-version: 3.14
           os: ubuntu-24.04-arm
+          no-httpx: false
           experimental: true
       fail-fast: false
     uses: ./.github/workflows/reusable-test.yml
@@ -57,6 +67,7 @@ jobs:
       os: ${{ matrix.os }}
       continue-on-error: ${{ matrix.experimental }}
       enable-cache: ${{ github.ref_type == 'tag' && 'false' || 'auto' }}
+      no-httpx: ${{ matrix.no-httpx }}
     secrets:
       codecov-token: ${{ secrets.CODECOV_TOKEN }}
 

.github/workflows/reusable-test.yml

@@ -17,6 +17,9 @@ on:
       enable-cache:
         type: string
         required: true
+      no-httpx:
+        type: boolean
+        required: true
     secrets:
       codecov-token:
         required: false
@@ -26,7 +29,9 @@ env:
 
 jobs:
   test:
-    name: Test Python ${{ inputs.python-version }} on ${{ inputs.os }}
+    name: Test Python ${{ inputs.python-version }} on ${{ inputs.os }}${{
+      inputs.no-httpx && ' no-httpx' || '' }}
+
     runs-on: ${{ inputs.os }}
     continue-on-error: ${{ inputs.continue-on-error }}
     env:
@@ -49,8 +54,16 @@ jobs:
       run: |
         uv run make pre-commit
     - name: Run unittests
+      if: ${{ ! inputs.no-httpx }}
+      env:
+        COLOR: 'yes'
+      run: |
+        uv run --extra=httpx make mototest
+    - name: Run unittests without httpx installed
+      if: ${{ inputs.no-httpx }}
       env:
         COLOR: 'yes'
+        HTTP_BACKEND: 'aiohttp'
       run: |
         uv run make mototest
     - name: Upload coverage to Codecov
@@ -60,7 +73,7 @@ jobs:
         token: ${{ secrets.codecov-token }}  # not required for public repos
         files: ./coverage.xml
         # yamllint disable-line rule:line-length
-        flags: unittests,os-${{ inputs.os }},python-${{ inputs.python-version }}  # optional
+        flags: unittests,os-${{ inputs.os }},python-${{ inputs.python-version }}${{ inputs.no-httpx && ',no-httpx' || '' }}  # optional
         name: codecov-umbrella  # optional
         fail_ci_if_error: true  # optional (default = false)
         verbose: true  # optional (default = false)

CHANGES.rst

@@ -10,6 +10,7 @@ Changes
 * patch ``_apply_request_trailer_checksum``, ``_get_provider_params`` and ``AIOWaiter.wait`` to add context feature information
 * patch ``AioSession._create_client`` to propagate context information. Created ``context.py`` to provide equivalent async context manager and async decorator functions
 * fixed test ``test_put_object_sha256`` as ``moto`` supports ``ChecksumSHA256``
+* Add experimental httpx support. The backend can be activated when creating a new session: ``session.create_client(..., config=AioConfig(http_session_cls=aiobotocore.httpxsession.HttpxSession))``. It's not fully tested and some features from aiohttp have not been ported, but feedback on what you're missing and bug reports are very welcome.
 
 2.22.0 (2025-04-29)
 ^^^^^^^^^^^^^^^^^^^

Makefile

@@ -1,6 +1,7 @@
 # Some simple testing tasks (sorry, UNIX only).
 
-# ?= conditional assign, so users can pass options on the CLI instead of manually editing this file
+# ?= is conditional assign, so users can pass options on the CLI instead of manually editing this file
+HTTP_BACKEND?='all'
 FLAGS?=
 
 pre-commit:
@@ -17,7 +18,7 @@ cov cover coverage: pre-commit
 	@echo "open file://`pwd`/htmlcov/index.html"
 
 mototest:
-	python -Wd -X tracemalloc=5 -X faulthandler -m pytest -vv -m "not localonly" -n auto --cov-report term --cov-report html --cov-report xml --cov=aiobotocore --cov=tests --log-cli-level=DEBUG $(FLAGS) aiobotocore tests
+	python -Wd -X tracemalloc=5 -X faulthandler -m pytest -vv -m "not localonly" -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
 
 clean:
 	rm -rf `find . -name __pycache__`

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,10 +19,26 @@
     asyncio.TimeoutError,
 ]
 
+
 botocore.retryhandler.EXCEPTION_MAP['GENERAL_CONNECTION_ERROR'].extend(
     _aiohttp_retryable_exceptions
 )
 
+if httpx is not None:
+    # See https://www.python-httpx.org/exceptions/#the-exception-hierarchy
+    # All child exceptions of TransportError, except ProxyError,
+    # UnsupportedProtocol and CloseError.
+    _httpx_retryable_exceptions = [
+        httpx.TimeoutException,
+        httpx.ProtocolError,
+        httpx.ConnectError,
+        httpx.ReadError,
+        httpx.WriteError,
+    ]
+    botocore.retryhandler.EXCEPTION_MAP['GENERAL_CONNECTION_ERROR'].extend(
+        _httpx_retryable_exceptions
+    )
+
 
 def _text(s, encoding='utf-8', errors='strict'):
     if isinstance(s, bytes):

aiobotocore/awsrequest.py

@@ -28,3 +28,14 @@ async def _text_prop(self):
     @property
     def text(self):
         return self._text_prop()
+
+
+class HttpxAWSResponse(AioAWSResponse):
+    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

aiobotocore/config.py

@@ -4,6 +4,17 @@
 from botocore.exceptions import ParamValidationError
 
 from aiobotocore.endpoint import DEFAULT_HTTP_SESSION_CLS
+from aiobotocore.httpxsession import HttpxSession
+
+# AWS has a 20 second idle timeout:
+#   https://web.archive.org/web/20150926192339/https://forums.aws.amazon.com/message.jspa?messageID=215367
+# and aiohttp default timeout is 30s so we set it to something
+# reasonable here
+DEFAULT_KEEPALIVE_TIMEOUT = 12
+
+TIMEOUT_ARGS = frozenset(
+    ('keepalive_timeout', 'write_timeout', 'pool_timeout')
+)
 
 
 class AioConfig(botocore.client.Config):
@@ -15,18 +26,16 @@ def __init__(
     ):
         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()
 
         if 'keepalive_timeout' not in self.connector_args:
-            # AWS has a 20 second idle timeout:
-            #   https://web.archive.org/web/20150926192339/https://forums.aws.amazon.com/message.jspa?messageID=215367
-            # and aiohttp default timeout is 30s so we set it to something
-            # reasonable here
-            self.connector_args['keepalive_timeout'] = 12
+            self.connector_args['keepalive_timeout'] = (
+                DEFAULT_KEEPALIVE_TIMEOUT
+            )
 
     def merge(self, other_config):
         # Adapted from parent class
@@ -35,13 +44,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'
@@ -51,12 +64,16 @@ def _validate_connector_args(connector_args):
                     raise ParamValidationError(
                         report=f'{k} value must be an int or None'
                     )
-            elif k == 'keepalive_timeout':
+            elif k in TIMEOUT_ARGS:
                 if v is not None and not isinstance(v, (float, int)):
                     raise ParamValidationError(
                         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'
@@ -72,6 +89,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

@@ -17,7 +17,12 @@
 from aiobotocore.httpchecksum import handle_checksum_body
 from aiobotocore.httpsession import AIOHTTPSession
 from aiobotocore.parsers import AioResponseParserFactory
-from aiobotocore.response import StreamingBody
+from aiobotocore.response import HttpxStreamingBody, StreamingBody
+
+try:
+    import httpx
+except ImportError:
+    httpx = None
 
 DEFAULT_HTTP_SESSION_CLS = AIOHTTPSession
 
@@ -50,8 +55,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'] = HttpxStreamingBody(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

aiobotocore/httpchecksum.py

@@ -13,7 +13,12 @@
 )
 
 from aiobotocore._helpers import resolve_awaitable
-from aiobotocore.response import StreamingBody
+from aiobotocore.response import HttpxStreamingBody, StreamingBody
+
+try:
+    import httpx
+except ImportError:
+    httpx = None
 
 
 class AioAwsChunkedWrapper(AwsChunkedWrapper):
@@ -109,6 +114,50 @@ def _validate_checksum(self):
             raise FlexibleChecksumError(error_msg=error_msg)
 
 
+# TODO: fix inheritance? read & _validate_checksum are the exact same as above
+# only diff is super class and how to call __init__
+class HttpxStreamingChecksumBody(HttpxStreamingBody):
+    def __init__(self, raw_stream, content_length, checksum, expected):
+        # HttpxStreamingbody doesn't use content_length
+        super().__init__(raw_stream)
+        self._checksum = checksum
+        self._expected = expected
+
+    # TODO: this class is largely (or possibly entirely) untested. The tests need to be
+    # more thoroughly rewritten wherever they directly create Streamingbody,
+    # StreamingChecksumBody, etc.
+
+    async def read(self, amt=None):
+        chunk = await super().read(amt=amt)
+        self._checksum.update(chunk)
+        if amt is None or (not chunk and amt > 0):
+            self._validate_checksum()
+        return chunk
+
+    async def readinto(self, b: bytearray):
+        chunk = await self.__wrapped__.content.read(len(b))
+        amount_read = len(chunk)
+        b[:amount_read] = chunk
+
+        if amount_read == len(b):
+            view = b
+        else:
+            view = memoryview(b)[:amount_read]
+
+        self._checksum.update(view)
+        if amount_read == 0 and len(b) > 0:
+            self._validate_checksum()
+        return amount_read
+
+    def _validate_checksum(self):
+        if self._checksum.digest() != base64.b64decode(self._expected):
+            error_msg = (
+                f"Expected checksum {self._expected} did not match calculated "
+                f"checksum: {self._checksum.b64digest()}"
+            )
+            raise FlexibleChecksumError(error_msg=error_msg)
+
+
 async def handle_checksum_body(
     http_response, response, context, operation_model
 ):
@@ -155,7 +204,11 @@ async def handle_checksum_body(
 def _handle_streaming_response(http_response, response, algorithm):
     checksum_cls = _CHECKSUM_CLS.get(algorithm)
     header_name = f"x-amz-checksum-{algorithm}"
-    return StreamingChecksumBody(
+    if httpx is not None and isinstance(http_response.raw, httpx.Response):
+        streaming_cls = HttpxStreamingChecksumBody
+    else:
+        streaming_cls = StreamingChecksumBody
+    return streaming_cls(
         http_response.raw,
         response["headers"].get("content-length"),
         checksum_cls(),

aiobotocore/httpsession.py

@@ -40,6 +40,7 @@
 from multidict import CIMultiDict
 
 import aiobotocore.awsrequest
+import aiobotocore.config
 from aiobotocore._endpoint_helpers import _IOBaseWrapper, _text
 
 
@@ -83,10 +84,9 @@ def __init__(
         self._timeout = timeout
         self._connector_args = connector_args
         if self._connector_args is None:
-            # AWS has a 20 second idle timeout:
-            #   https://web.archive.org/web/20150926192339/https://forums.aws.amazon.com/message.jspa?messageID=215367
-            # aiohttp default timeout is 30s so set something reasonable here
-            self._connector_args = dict(keepalive_timeout=12)
+            self._connector_args = dict(
+                keepalive_timeout=aiobotocore.config.DEFAULT_KEEPALIVE_TIMEOUT
+            )
 
         self._max_pool_connections = max_pool_connections
         self._socket_options = socket_options