feat: public async transport

CHANGELOG.md

@@ -9,7 +9,7 @@
 * (Major): `initialize_client()` no longer raises when registration fails. A failed registration is logged and the client keeps running, the same way for both clients.
 * (Major): Upgraded to yggdrasil-engine 3, new engine built with PyO3 for better performance.
 * (Minor): Feature fetches are retried the same way by both clients: on a 500, 502 or 504, or when no response arrives, with a short randomised backoff between attempts (up to about 0.5s, 1s and 2s). Before, retries were immediate.
-* (Minor): `UnleashClient` takes `transport=` to choose how its requests are sent: a `RequestsTransport` with your own `requests.Session`, for a proxy, TLS settings or connection reuse, or any object with an `execute(Request) -> Response` method, for another HTTP library. See [custom HTTP transport](./README.md#custom-http-transport).
+* (Minor): `UnleashClient` and `AsyncUnleashClient` take `transport=` to choose how their requests are sent: a `RequestsTransport` with custom `requests.Session`, an `AiohttpTransport` with custom `aiohttp.ClientSession`, for a proxy, TLS settings or connection reuse, or any object with an `execute(Request) -> Response` method, for another HTTP library. See [custom HTTP transport](./README.md#custom-http-transport).
 * (Minor): New `FETCH_COMPLETED` event, emitted after every feature fetch with how it ended: `CHANGED`, `NOT_CHANGED`, `UNAUTHORIZED`, `RATE_LIMITED`, `UNAVAILABLE` or `INVALID_RESPONSE`.
 * (Minor): `feature_definitions()` reports whether each feature is enabled in the environment.
 * (Minor): New `evaluate_all_toggles()` evaluates every feature for one context in a single call, without recording usage metrics.

README.md

@@ -149,7 +149,7 @@ Settings can't be changed after the client is created.
 | scheduler_executor     | Name of the executor to use on your custom `scheduler`. Required with `scheduler`. Sync client only.                                      | None                |
 | multiple_instance_mode | How to handle multiple client instances (`BLOCK`, `WARN`, `SILENTLY_ALLOW`).                                                              | `WARN`              |
 | event_callback         | Function that receives impression and lifecycle events. See [events and impression data](#events-and-impression-data).                    | None                |
-| transport              | How requests are sent. Can't be combined with `http` (which will be removed). Sync client only yet. See [custom HTTP transport](#custom-http-transport).    | `RequestsTransport` |
+| transport              | How requests are sent. Can't be combined with `http` (which will be removed). See [custom HTTP transport](#custom-http-transport).    | `RequestsTransport` or `AiohttpTransport` |
 
 #### `IdentityOptions`
 
@@ -471,12 +471,12 @@ client = UnleashClient(
 
 ### Custom HTTP transport
 
-By default, `UnleashClient` sends its requests with `requests`, on a new session for every request. Pass `transport=` to choose how requests are sent:
+By default, `UnleashClient` sends its requests with `requests`, on a new session for every request, and `AsyncUnleashClient` with one `aiohttp` session. Pass `transport=` to choose how requests are sent:
 
-- `RequestsTransport(session)` to keep `requests` but configure the session yourself;
-- any object with an `execute(request) -> Response` method, to use another HTTP libraryl.
+- `RequestsTransport(session)` or `AiohttpTransport(session)` to keep the default library but configure the session;
+- any object with an `execute(request) -> Response` method, to use another HTTP library. For `AsyncUnleashClient`, `execute` is `async`.
 
-#### Your own `requests` session
+#### Custom `requests` session
 
 The session is used for every request and never closed by the client: close it yourself after `client.destroy()`.
 
@@ -515,6 +515,39 @@ A session also keeps connections open between polls, instead of opening a new on
 
 `RequestsTransport` takes `headers` and `timeout` with or without a session. The SDK's own headers take precedence over custom ones, and an `Authorization` header raises `ValueError`: pass the API token as `api_key`.
 
+#### Custom `aiohttp` session
+
+The same for `AsyncUnleashClient`, with an `aiohttp.ClientSession`. Create it inside a coroutine, since aiohttp needs a running event loop, and close it after the client:
+
+```python
+import ssl
+import aiohttp
+from UnleashClient import AiohttpTransport
+from UnleashClient.clients.async_unleash_client import AsyncUnleashClient
+
+# examples
+tls = ssl.create_default_context(cafile="/etc/ssl/certs/internal-ca.pem")
+tls.load_cert_chain("/etc/unleash/client.crt", "/etc/unleash/client.key")
+
+async def main():
+    async with aiohttp.ClientSession(
+        trust_env=True,
+        connector=aiohttp.TCPConnector(ssl=tls),
+        trace_configs=[my_trace_config],  # e.g. aiohttp tracing for OpenTelemetry
+    ) as session:
+        async with AsyncUnleashClient(
+            url="https://YOUR-API-URL",
+            app_name="my-python-app",
+            api_key="<API token>",
+            transport=AiohttpTransport(session, headers={"X-Team": "checkout"}, timeout=10),
+        ) as client:
+            ...
+```
+
+`AiohttpTransport`'s `timeout` applies to every request, in place of the session's own. Every status reaches the client as a response, even on a session created with `raise_for_status=True`.
+
+An `AiohttpTransport()` without a session opens its own; if you create one and pass it in, close it with `await transport.aclose()` after the client.
+
 #### Another HTTP library
 
 A transport receives a `Request` (`method`, `url`, `headers`, `params`, `body`) and returns a `Response` (`status`, `headers`, `body` as bytes). Every status is a response: raise only when no response arrived, such as on a connection error or a timeout. The client logs the error and reports a failed request. For example, with [httpx](https://www.python-httpx.org/):
@@ -546,6 +579,38 @@ with httpx.Client(timeout=30) as http_client:
     )
 ```
 
+For `AsyncUnleashClient`, `execute` is `async`. The same with `httpx.AsyncClient`:
+
+```python
+import httpx
+from UnleashClient import Request, Response
+from UnleashClient.clients.async_unleash_client import AsyncUnleashClient
+
+class HttpxAsyncTransport:
+    def __init__(self, client: httpx.AsyncClient) -> None:
+        self._client = client
+
+    async def execute(self, request: Request) -> Response:
+        resp = await self._client.request(
+            request.method,
+            request.url,
+            headers=dict(request.headers),
+            params=dict(request.params),
+            content=request.body,
+        )
+        return Response(status=resp.status_code, headers=resp.headers, body=resp.content)
+
+async def main():
+    async with httpx.AsyncClient(timeout=30) as http_client:
+        async with AsyncUnleashClient(
+            url="https://YOUR-API-URL",
+            app_name="my-python-app",
+            api_key="<API token>",
+            transport=HttpxAsyncTransport(http_client),
+        ) as client:
+            ...
+```
+
 The response headers can be a plain `dict`: the client looks them up case-insensitively.
 
 #### Tests without a server
@@ -580,11 +645,33 @@ def test_new_checkout_is_on(tmp_path):
         assert client.is_enabled("new-checkout")
 ```
 
-Both examples are in [`examples/`](examples/) and run as part of the SDK's tests.
+For `AsyncUnleashClient`, make `execute` `async` and the test a coroutine, here with [pytest-asyncio](https://pytest-asyncio.readthedocs.io/):
+
+```python
+import pytest
+from UnleashClient.clients.async_unleash_client import AsyncUnleashClient
+
+
+class FakeAsyncTransport(FakeTransport):
+    async def execute(self, request: Request) -> Response:
+        return super().execute(request)
+
+@pytest.mark.asyncio
+async def test_new_checkout_is_on(tmp_path):
+    transport = FakeAsyncTransport(flag("new-checkout", True))
+
+    async with AsyncUnleashClient(
+        "http://unleash.test/api", "my-app-tests",
+        transport=transport, cache_directory=str(tmp_path),
+    ) as client:
+        assert client.is_enabled("new-checkout")
+```
+
+All four examples are in [`examples/`](examples/) and run as part of the SDK's tests.
 
 #### Things to know
 
-- **Don't retry in your transport.** The client already retries failed feature fetches, with a backoff. A session with a retrying adapter mounted logs a warning, because failed fetches would be retried twice over.
+- **Don't retry in your transport.** The client already retries failed feature fetches, with a backoff. A `requests` session with a retrying adapter mounted logs a warning, because failed fetches would be retried twice over. aiohttp has no retry setting, but resends a GET once by itself when the server drops a reused connection; that's expected, and not a retry to turn off.
 - **Don't log request headers.** They include your API token.
 - **Don't add an HTTP cache.** The client already sends `If-None-Match` and handles `304 Not Modified`. A caching session (such as CacheControl or requests-cache) turns each 304 into its cached 200, so the client reloads unchanged flags on every poll. The flags stay correct.
 - **Create the session in the process that uses it.** With a forking server, such as gunicorn with `--preload` or uWSGI without `lazy-apps`, build the session together with the client in each worker. Forked processes must not share open connections. See [running in multi-process setups](#running-in-multi-process-setups).

UnleashClient/__init__.py

@@ -13,9 +13,10 @@
     PollingOptions,
     StreamingOptions,
 )
-from .transport import Request, RequestsTransport, Response, Transport
+from .transport import AsyncTransport, Request, RequestsTransport, Response, Transport
 
 if TYPE_CHECKING:
+    from ._aiohttp_transport import AiohttpTransport
     from .clients.async_unleash_client import AsyncUnleashClient
 
 
@@ -24,4 +25,8 @@ def __getattr__(name: str):
         from .clients.async_unleash_client import AsyncUnleashClient  # noqa: PLC0415
 
         return AsyncUnleashClient
+    if name == "AiohttpTransport":
+        from ._aiohttp_transport import AiohttpTransport  # noqa: PLC0415
+
+        return AiohttpTransport
     raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

UnleashClient/_aiohttp_transport.py

@@ -3,14 +3,15 @@
 and returns the response.
 """
 
-from typing import Optional
+from typing import Mapping, Optional
 
 from UnleashClient.constants import REQUEST_TIMEOUT
 from UnleashClient.errors import AlreadyClosedError
-from UnleashClient.transport import Request, Response
+from UnleashClient.transport import Request, Response, _reject_authorization
 
 try:
     import aiohttp
+    from multidict import CIMultiDict
 except ImportError as exc:  # pragma: no cover
     raise ImportError(
         "AsyncUnleashClient requires aiohttp, which is not installed. "
@@ -22,21 +23,47 @@ class AiohttpTransport:
     """
     Sends requests with ``aiohttp``, on one session reused across requests.
 
-    The session is opened on the first request, since aiohttp needs a running
-    event loop to create one, and closed by :meth:`aclose`. A closed transport
-    stays closed.
 
-    :param timeout: Seconds to wait for the connection and for each read.
+    Example::
+
+        async with aiohttp.ClientSession(trust_env=True) as session:
+            async with AsyncUnleashClient(
+                url, app_name, transport=AiohttpTransport(session)
+            ) as client:
+                ...
+
+    :param session: Session to send every request on. It is never closed: its
+                    creator closes it, after the client. Without one, the
+                    transport opens its own on the first request, and
+                    :meth:`aclose` closes it.
+    :param headers: Extra headers sent with every request. The SDK's own headers
+                    take precedence over them.
+    :param timeout: Seconds to wait for the connection and for each read. Applies
+                    to every request, in place of the session's own timeout.
+    :raises ValueError: If ``headers`` contains an ``Authorization`` header: pass
+                        the API key as ``api_key`` to the client.
     """
 
-    def __init__(self, *, timeout: float = REQUEST_TIMEOUT) -> None:
+    def __init__(
+        self,
+        session: Optional["aiohttp.ClientSession"] = None,
+        *,
+        headers: Optional[Mapping[str, str]] = None,
+        timeout: float = REQUEST_TIMEOUT,
+    ) -> None:
+        self._user_session = session
+        self._headers = dict(headers or {})
         self._timeout = timeout
         self._session: Optional["aiohttp.ClientSession"] = None
         self._is_closed: bool = False
 
+        _reject_authorization(self._headers, "AiohttpTransport")
+
     def _get_session(self) -> "aiohttp.ClientSession":
         if self._is_closed:
             raise AlreadyClosedError()
+        if self._user_session is not None:
+            return self._user_session
         if self._session is None:
             self._session = aiohttp.ClientSession()
         return self._session
@@ -49,21 +76,25 @@ def _client_timeout(self) -> "aiohttp.ClientTimeout":
         )
 
     async def execute(self, request: Request) -> Response:
+        headers = CIMultiDict(self._headers)
+        headers.update(request.headers)
         async with self._get_session().request(
             request.method,
             request.url,
-            headers=dict(request.headers),
+            headers=headers,
             params=dict(request.params) or None,
             data=request.body,
             timeout=self._client_timeout(),
+            raise_for_status=False,
         ) as resp:
             body = await resp.read()
             return Response(status=resp.status, headers=resp.headers, body=body)
 
     async def aclose(self) -> None:
         """
-        Close the session. Every later request raises
-        :class:`AlreadyClosedError`. Closing a closed transport does nothing.
+        Put the transport out of service, and close its session if the transport
+        opened it. Every later request raises :class:`AlreadyClosedError`.
+        Closing a closed transport does nothing.
         """
         self._is_closed = True
         session, self._session = self._session, None

UnleashClient/_async_unleash_api_client.py

@@ -14,6 +14,7 @@
     _emit_fetch_completed,
     _failure_status,
     _fetched_status,
+    _header,
     _log_resp_info,
     _normalized_url,
     _text,
@@ -141,7 +142,7 @@ async def _fetch_features(
                     "Unleash Client feature fetch failed!"
                 )
 
-            fetched_etag = resp.headers.get("etag", "")
+            fetched_etag = _header(resp, "etag")
 
             if resp.status == 304:
                 return (

UnleashClient/clients/async_unleash_client.py

@@ -10,6 +10,7 @@
 
 from yggdrasil_engine.engine import UnleashEngine
 
+from UnleashClient._aiohttp_transport import aiohttp
 from UnleashClient._async_metrics import _AsyncMetricsReporter
 from UnleashClient._async_scheduler import _AsyncScheduler
 from UnleashClient._async_unleash_api_client import _AsyncUnleashApiClient
@@ -37,6 +38,7 @@
 from UnleashClient.constants import ETAG, METRIC_LAST_SENT_TIME
 from UnleashClient.events import BaseEvent
 from UnleashClient.impact_metrics import ImpactMetrics
+from UnleashClient.transport import AsyncTransport, _check_transport
 from UnleashClient.utils import LOGGER, InstanceAllowType
 
 
@@ -78,6 +80,7 @@ class AsyncUnleashClient:
     :param cache_directory: Location of the cache directory.
     :param multiple_instance_mode: Determines how multiple instances being instantiated is handled by the SDK.  Defaults to InstanceAllowType.WARN.
     :param event_callback: Function to call if impression events are enabled.  Called on a dedicated background thread.
+    :param transport: Sends the requests to the server, see :class:`~UnleashClient.transport.AsyncTransport`.  Pass an :class:`AiohttpTransport` with your own ``aiohttp.ClientSession`` for a proxy, TLS settings or tracing.  The client never closes it.  Can't be combined with ``http``.
     """
 
     def __init__(
@@ -96,7 +99,16 @@ def __init__(
         cache_directory: Optional[str] = None,
         multiple_instance_mode: InstanceAllowType = InstanceAllowType.WARN,
         event_callback: Optional[Callable[[BaseEvent], None]] = None,
+        transport: Optional[AsyncTransport] = None,
     ) -> None:
+        if transport is not None:
+            _check_transport(
+                transport,
+                http_given=http is not None,
+                is_async=True,
+                session_type=aiohttp.ClientSession,
+                wrapper_name="AiohttpTransport",
+            )
         if isinstance(mode, StreamingOptions):
             raise ValueError(
                 "AsyncUnleashClient does not support streaming yet. "
@@ -143,7 +155,7 @@ def __init__(
             events=self._event_dispatcher,
         )
         self._api: _AsyncUnleashApiClient = _AsyncUnleashApiClient(
-            self._config, self._headers, self._event_dispatcher
+            self._config, self._headers, self._event_dispatcher, transport=transport
         )
         self._metrics_scheduler: _AsyncScheduler = _AsyncScheduler()
         self._metrics: _AsyncMetricsReporter = _AsyncMetricsReporter(