Ensure that all async generators are explicitly closed

.github/workflows/test-suite.yml

@@ -14,11 +14,11 @@ jobs:
 
     strategy:
       matrix:
-        python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"]
+        python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"]
 
     steps:
       - uses: "actions/checkout@v4"
-      - uses: "actions/setup-python@v4"
+      - uses: "actions/setup-python@v5"
         with:
           python-version: "${{ matrix.python-version }}"
           allow-prereleases: true

CHANGELOG.md

@@ -4,6 +4,11 @@ All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 
+## [Unreleased]
+
+- Drop Python 3.8 support
+- Explicitly close all async generators to ensure predictable behavior
+
 ## Version 1.0.9 (April 24th, 2025)
 
 - Resolve https://github.com/advisories/GHSA-vqfr-h8mv-ghfj with h11 dependency update. (#1008)

httpcore/_async/connection_pool.py

@@ -4,6 +4,8 @@
 import sys
 import types
 import typing
+from collections.abc import AsyncGenerator
+from contextlib import AsyncExitStack
 
 from .._backends.auto import AutoBackend
 from .._backends.base import SOCKET_OPTION, AsyncNetworkBackend
@@ -398,13 +400,14 @@ def __init__(
         self._pool = pool
         self._closed = False
 
-    async def __aiter__(self) -> typing.AsyncIterator[bytes]:
-        try:
-            async for part in self._stream:
-                yield part
-        except BaseException as exc:
-            await self.aclose()
-            raise exc from None
+    async def __aiter__(self) -> AsyncGenerator[bytes]:
+        async with AsyncExitStack() as stack:
+            iterator = self._stream.__aiter__()
+            if hasattr(iterator, "aclose"):
+                stack.push_async_callback(iterator.aclose)
+
+            async for chunk in iterator:
+                yield chunk
 
     async def aclose(self) -> None:
         if not self._closed:

httpcore/_async/http11.py

@@ -6,6 +6,7 @@
 import time
 import types
 import typing
+from collections.abc import AsyncGenerator
 
 import h11
 
@@ -20,6 +21,7 @@
 from .._models import Origin, Request, Response
 from .._synchronization import AsyncLock, AsyncShieldCancellation
 from .._trace import Trace
+from .._utils import aclosing
 from .interfaces import AsyncConnectionInterface
 
 logger = logging.getLogger("httpcore.http11")
@@ -193,9 +195,7 @@ async def _receive_response_headers(
 
         return http_version, event.status_code, event.reason, headers, trailing_data
 
-    async def _receive_response_body(
-        self, request: Request
-    ) -> typing.AsyncIterator[bytes]:
+    async def _receive_response_body(self, request: Request) -> AsyncGenerator[bytes]:
         timeouts = request.extensions.get("timeout", {})
         timeout = timeouts.get("read", None)
 
@@ -327,12 +327,15 @@ def __init__(self, connection: AsyncHTTP11Connection, request: Request) -> None:
         self._request = request
         self._closed = False
 
-    async def __aiter__(self) -> typing.AsyncIterator[bytes]:
+    async def __aiter__(self) -> AsyncGenerator[bytes]:
         kwargs = {"request": self._request}
         try:
             async with Trace("receive_response_body", logger, self._request, kwargs):
-                async for chunk in self._connection._receive_response_body(**kwargs):
-                    yield chunk
+                async with aclosing(
+                    self._connection._receive_response_body(**kwargs)
+                ) as body:
+                    async for chunk in body:
+                        yield chunk
         except BaseException as exc:
             # If we get an exception while streaming the response,
             # we want to close the response (and possibly the connection)

httpcore/_async/http2.py

@@ -5,6 +5,7 @@
 import time
 import types
 import typing
+from collections.abc import AsyncGenerator
 
 import h2.config
 import h2.connection
@@ -308,7 +309,7 @@ async def _receive_response(
 
     async def _receive_response_body(
         self, request: Request, stream_id: int
-    ) -> typing.AsyncIterator[bytes]:
+    ) -> AsyncGenerator[bytes]:
         """
         Iterator that returns the bytes of the response body for a given stream ID.
         """
@@ -568,7 +569,7 @@ def __init__(
         self._stream_id = stream_id
         self._closed = False
 
-    async def __aiter__(self) -> typing.AsyncIterator[bytes]:
+    async def __aiter__(self) -> AsyncGenerator[bytes]:
         kwargs = {"request": self._request, "stream_id": self._stream_id}
         try:
             async with Trace("receive_response_body", logger, self._request, kwargs):

httpcore/_async/interfaces.py

@@ -2,6 +2,7 @@
 
 import contextlib
 import typing
+from collections.abc import AsyncGenerator
 
 from .._models import (
     URL,
@@ -58,7 +59,7 @@ async def stream(
         headers: HeaderTypes = None,
         content: bytes | typing.AsyncIterator[bytes] | None = None,
         extensions: Extensions | None = None,
-    ) -> typing.AsyncIterator[Response]:
+    ) -> AsyncGenerator[Response]:
         # Strict type checking on our parameters.
         method = enforce_bytes(method, name="method")
         url = enforce_url(url, name="url")

httpcore/_models.py

@@ -4,6 +4,11 @@
 import ssl
 import typing
 import urllib.parse
+from collections.abc import AsyncGenerator
+from contextlib import AsyncExitStack
+from inspect import isasyncgen
+
+from ._utils import aclosing
 
 # Functions for typechecking...
 
@@ -151,7 +156,7 @@ def __init__(self, content: bytes) -> None:
     def __iter__(self) -> typing.Iterator[bytes]:
         yield self._content
 
-    async def __aiter__(self) -> typing.AsyncIterator[bytes]:
+    async def __aiter__(self) -> AsyncGenerator[bytes]:
         yield self._content
 
     def __repr__(self) -> str:
@@ -463,10 +468,11 @@ async def aread(self) -> bytes:
                 "You should use 'response.read()' instead."
             )
         if not hasattr(self, "_content"):
-            self._content = b"".join([part async for part in self.aiter_stream()])
+            async with aclosing(self.aiter_stream()) as parts:
+                self._content = b"".join([part async for part in parts])
         return self._content
 
-    async def aiter_stream(self) -> typing.AsyncIterator[bytes]:
+    async def aiter_stream(self) -> AsyncGenerator[bytes]:
         if not isinstance(self.stream, typing.AsyncIterable):  # pragma: nocover
             raise RuntimeError(
                 "Attempted to stream an synchronous response using 'async for ... in "
@@ -479,8 +485,13 @@ async def aiter_stream(self) -> typing.AsyncIterator[bytes]:
                 "more than once."
             )
         self._stream_consumed = True
-        async for chunk in self.stream:
-            yield chunk
+        async with AsyncExitStack() as stack:
+            iterator = self.stream.__aiter__()
+            if isasyncgen(iterator):
+                stack.push_async_callback(iterator.aclose)
+
+            async for chunk in iterator:
+                yield chunk
 
     async def aclose(self) -> None:
         if not isinstance(self.stream, typing.AsyncIterable):  # pragma: nocover

httpcore/_sync/connection_pool.py

@@ -4,6 +4,8 @@
 import sys
 import types
 import typing
+from collections.abc import Generator
+from contextlib import ExitStack
 
 from .._backends.sync import SyncBackend
 from .._backends.base import SOCKET_OPTION, NetworkBackend
@@ -398,13 +400,14 @@ def __init__(
         self._pool = pool
         self._closed = False
 
-    def __iter__(self) -> typing.Iterator[bytes]:
-        try:
-            for part in self._stream:
-                yield part
-        except BaseException as exc:
-            self.close()
-            raise exc from None
+    def __iter__(self) -> Generator[bytes]:
+        with ExitStack() as stack:
+            iterator = self._stream.__iter__()
+            if hasattr(iterator, "close"):
+                stack.callback(iterator.close)
+
+            for chunk in iterator:
+                yield chunk
 
     def close(self) -> None:
         if not self._closed:

httpcore/_sync/http11.py

@@ -6,6 +6,7 @@
 import time
 import types
 import typing
+from collections.abc import Generator
 
 import h11
 
@@ -20,6 +21,7 @@
 from .._models import Origin, Request, Response
 from .._synchronization import Lock, ShieldCancellation
 from .._trace import Trace
+from contextlib import closing
 from .interfaces import ConnectionInterface
 
 logger = logging.getLogger("httpcore.http11")
@@ -193,9 +195,7 @@ def _receive_response_headers(
 
         return http_version, event.status_code, event.reason, headers, trailing_data
 
-    def _receive_response_body(
-        self, request: Request
-    ) -> typing.Iterator[bytes]:
+    def _receive_response_body(self, request: Request) -> Generator[bytes]:
         timeouts = request.extensions.get("timeout", {})
         timeout = timeouts.get("read", None)
 
@@ -327,12 +327,15 @@ def __init__(self, connection: HTTP11Connection, request: Request) -> None:
         self._request = request
         self._closed = False
 
-    def __iter__(self) -> typing.Iterator[bytes]:
+    def __iter__(self) -> Generator[bytes]:
         kwargs = {"request": self._request}
         try:
             with Trace("receive_response_body", logger, self._request, kwargs):
-                for chunk in self._connection._receive_response_body(**kwargs):
-                    yield chunk
+                with closing(
+                    self._connection._receive_response_body(**kwargs)
+                ) as body:
+                    for chunk in body:
+                        yield chunk
         except BaseException as exc:
             # If we get an exception while streaming the response,
             # we want to close the response (and possibly the connection)

httpcore/_sync/http2.py

@@ -5,6 +5,7 @@
 import time
 import types
 import typing
+from collections.abc import Generator
 
 import h2.config
 import h2.connection
@@ -308,7 +309,7 @@ def _receive_response(
 
     def _receive_response_body(
         self, request: Request, stream_id: int
-    ) -> typing.Iterator[bytes]:
+    ) -> Generator[bytes]:
         """
         Iterator that returns the bytes of the response body for a given stream ID.
         """
@@ -568,7 +569,7 @@ def __init__(
         self._stream_id = stream_id
         self._closed = False
 
-    def __iter__(self) -> typing.Iterator[bytes]:
+    def __iter__(self) -> Generator[bytes]:
         kwargs = {"request": self._request, "stream_id": self._stream_id}
         try:
             with Trace("receive_response_body", logger, self._request, kwargs):

httpcore/_sync/interfaces.py

@@ -2,6 +2,7 @@
 
 import contextlib
 import typing
+from collections.abc import Generator
 
 from .._models import (
     URL,
@@ -58,7 +59,7 @@ def stream(
         headers: HeaderTypes = None,
         content: bytes | typing.Iterator[bytes] | None = None,
         extensions: Extensions | None = None,
-    ) -> typing.Iterator[Response]:
+    ) -> Generator[Response]:
         # Strict type checking on our parameters.
         method = enforce_bytes(method, name="method")
         url = enforce_url(url, name="url")

httpcore/_utils.py

@@ -4,6 +4,27 @@
 import socket
 import sys
 
+if sys.version_info >= (3, 10):  # pragma: no cover
+    from contextlib import aclosing as aclosing
+else:  # pragma: no cover
+    from contextlib import AbstractAsyncContextManager
+    from typing import Any, Awaitable, Protocol, TypeVar
+
+    class _SupportsAclose(Protocol):
+        def aclose(self) -> Awaitable[object]: ...
+
+    _SupportsAcloseT = TypeVar("_SupportsAcloseT", bound=_SupportsAclose)
+
+    class aclosing(AbstractAsyncContextManager[_SupportsAcloseT, None]):
+        def __init__(self, thing: _SupportsAcloseT) -> None:
+            self.thing = thing
+
+        async def __aenter__(self) -> _SupportsAcloseT:
+            return self.thing
+
+        async def __aexit__(self, *exc_info: Any) -> None:
+            await self.thing.aclose()
+
 
 def is_socket_readable(sock: socket.socket | None) -> bool:
     """

pyproject.toml

@@ -7,7 +7,7 @@ name = "httpcore"
 dynamic = ["readme", "version"]
 description = "A minimal low-level HTTP client."
 license = "BSD-3-Clause"
-requires-python = ">=3.8"
+requires-python = ">=3.9"
 authors = [
     { name = "Tom Christie", email = "[email protected]" },
 ]
@@ -26,6 +26,7 @@ classifiers = [
     "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
+    "Programming Language :: Python :: 3.13",
     "Topic :: Internet :: WWW/HTTP",
 ]
 dependencies = [
@@ -96,7 +97,7 @@ filterwarnings = ["error"]
 
 [tool.coverage.run]
 omit = [
-    "venv/*", 
+    "venv/*",
     "httpcore/_sync/*"
 ]
 include = ["httpcore/*", "tests/*"]

requirements.txt

@@ -15,7 +15,7 @@ twine==6.1.0
 # Tests & Linting
 coverage[toml]==7.5.4
 ruff==0.5.0
-mypy==1.10.1
+mypy==1.14.1
 trio-typing==0.10.0
 pytest==8.2.2
 pytest-httpbin==2.0.0