Reuse HTTP connections and add close() to both clients

CONTRIBUTING.md

@@ -104,6 +104,19 @@ project sees an installed permit, and fails while `permit/_sync_types.pyi` is ou
 (see [Regenerating the sync stubs](#regenerating-the-sync-stubs)). The `mypy` pre-commit
 hook type-checks the SDK itself, strictly and with the pydantic plugin (see [Setup](#setup)).
 
+### Connection reuse
+
+pytest-httpserver closes each connection after its response, so the tests of how the
+clients keep and close their connections (`tests/test_async_session_lifecycle.py` and
+`tests/test_sync_lifecycle.py`) use `tests/keepalive_server.py`, a local HTTP/1.1 server
+that keeps every connection open and counts the connections it accepted and those that were
+closed. The benchmark runs on it too: it times sequential `check()` calls of the async and
+the blocking client, and prints how many connections each opened.
+
+```sh
+uv run --locked python -m tests.benchmark_connection_reuse --calls 500
+```
+
 ### The migration skill's tests
 
 `skills/tests` checks `MIGRATION.md` and the permit-python-3-migration skill against each

README.md

@@ -21,6 +21,74 @@ every breaking change, who it affects and what to change. To have an AI agent su
 do the upgrade, use the
 [permit-python-3-migration skill](https://github.com/permitio/permit-python/tree/main/skills/permit-python-3-migration).
 
+## Connections
+
+Both clients keep the HTTP connections they open and reuse them for their next requests, so a
+request does not pay for a new connection, and a TLS handshake, each time. A client keeps one
+pool of connections for the Permit API and one for the PDP, opened by its first request.
+
+### The async client
+
+`permit.Permit` keeps its pools per event loop it is used on, each opened by the first
+request from that loop.
+
+```py
+async with Permit(token="<YOUR_API_KEY>") as permit:
+    allowed = await permit.check("alice", "read", "document")
+```
+
+- `await permit.close()` closes the connections, as leaving the `async with` block does.
+  Calling it again does nothing more, and the client stays usable: a request sent after it
+  opens new connections.
+- A client you never close leaves nothing open when its loop shuts down through
+  `asyncio.run()`, `asyncio.Runner` or anything else that shuts down the loop's async
+  generators before closing it: the client's connections on that loop are closed then. A
+  client that is garbage collected while its loop runs closes its connections on that
+  loop. As the interpreter exits, the client closes what is still open, so aiohttp reports
+  no unclosed session.
+- If you drive an event loop yourself, run `await permit.close()` on it before you close it.
+  A loop closed with `loop.close()` alone cannot close its connections any more: they stay
+  open until the client's next request, from any loop, lets the garbage collector free
+  them, and Python reports each one with a `ResourceWarning`. Python's default warning
+  filters hide it, but a test suite that turns warnings into errors, such as pytest with
+  `filterwarnings = error`, fails on it.
+- Close the client once no request is in flight: a request in flight when `close()` runs
+  fails.
+
+### The blocking client
+
+`permit.sync.Permit` runs its calls on an event loop in a background daemon thread of its
+own, which it starts on its first call. Calls from every thread that uses the client are
+handed to that thread and waited for, so they share the client's connections.
+
+```py
+from permit.sync import Permit
+
+with Permit(token="<YOUR_API_KEY>") as permit:
+    allowed = permit.check("alice", "read", "document")
+```
+
+- `permit.close()` waits for the calls other threads have in flight, closes the connections
+  and stops the thread, as leaving the `with` block does. A call or a `close()` another
+  thread makes meanwhile waits for it to finish. Calling it again does nothing more, and the
+  client stays usable: its next call starts a new thread and opens new connections.
+- A client you never close is cleaned up when it is garbage collected, or as the
+  interpreter exits. The thread never holds up the exit.
+- Do not call the blocking client from code that runs on its own background thread, such as
+  a callback scheduled on its loop: such a call, and `close()`, raise `RuntimeError` rather
+  than wait for themselves.
+
+### Both clients
+
+- With `proxy_facts_via_pdp` on, `wait_for_sync()` yields a client that uses the connections
+  of the client it is called on, and on the blocking client its thread too. That client's
+  `close()` closes them; the yielded one's `close()` does nothing. With it off, the default,
+  `wait_for_sync()` logs a warning and yields the client itself, whose `close()` closes them.
+- A child process made by `fork()` leaves the connections it inherits to its parent, and
+  opens its own; the blocking client starts a thread of its own in the child.
+- The number of connections open at once is not capped, as before. An idle connection is
+  closed after aiohttp's keep-alive timeout of 15 seconds.
+
 ## Groups
 
 `permit.api.groups` manages groups. A group is a resource instance, of the `group` resource

permit/api/base.py

@@ -1,9 +1,11 @@
 from typing import TYPE_CHECKING, Any, TypeVar, cast, overload
 
-import aiohttp
 from aiohttp import ClientTimeout
+from multidict import CIMultiDict
+from yarl import URL
 
 from permit.api.encoders import jsonable_encoder
+from permit.utils.http_sessions import LoopSessions
 from permit.utils.pydantic_version import PYDANTIC_VERSION
 from permit.utils.sdk_logger import sdk_logger
 
@@ -56,16 +58,100 @@ class Config:
     )
 
 
+# What a SimpleHttpClient's client_config may set. The other options of an aiohttp session
+# cannot be set per client, since the client sends its requests through shared sessions.
+_CLIENT_CONFIG_KEYS = frozenset({"base_url", "headers", "timeout"})
+
+
+def _session_base_url(base_url: str | URL) -> URL:
+    """``base_url`` as ``aiohttp.ClientSession(base_url=...)`` reads it, raising what it raises.
+
+    Raises:
+        ValueError: If ``base_url`` has no scheme or host, or its path does not end with "/".
+    """
+    if isinstance(base_url, URL):
+        url = base_url
+    else:
+        url = URL(base_url)
+        url.origin()  # raises ValueError for a URL without a scheme and a host
+    if not url.path.endswith("/"):
+        msg = "base_url must have a trailing '/'"
+        raise ValueError(msg)
+    return url
+
+
 class SimpleHttpClient:
-    """wraps aiohttp client to reduce boilerplace."""
+    """Sends requests to one endpoint and parses their JSON responses.
+
+    The requests go through ``sessions``, which keep their connections open for the next
+    request. Everything else a request carries comes from this client and the call itself,
+    so the sessions can serve every client of an SDK client.
+
+    Args:
+        client_config: Optional request settings: ``base_url``, the server a relative request
+            URL is resolved against, as ``aiohttp.ClientSession(base_url=...)`` resolves it;
+            ``headers``, sent with every request; and ``timeout``, an
+            ``aiohttp.ClientTimeout``.
+        base_url: The endpoint's path, put before the URL of every request.
+        timeout: The total timeout of each request in seconds, in place of
+            ``client_config["timeout"]``.
+        sessions: The sessions to send the requests through. Without them, the client has
+            sessions of its own.
+
+    Raises:
+        TypeError: If ``client_config`` has a key other than those above.
+    """
 
     def __init__(
-        self, client_config: dict[str, Any], base_url: str = "", timeout: int | None = None
+        self,
+        client_config: dict[str, Any],
+        base_url: str = "",
+        timeout: int | None = None,
+        *,
+        sessions: LoopSessions | None = None,
     ) -> None:
-        self._client_config = client_config
+        unsupported = sorted(set(client_config) - _CLIENT_CONFIG_KEYS)
+        if unsupported:
+            msg = (
+                f"SimpleHttpClient does not take the client_config keys {unsupported}: "
+                f"it sets only {sorted(_CLIENT_CONFIG_KEYS)} on its requests."
+            )
+            raise TypeError(msg)
+        self._server_url: str | URL | None = client_config.get("base_url")
+        self._headers: dict[str, str] | None = client_config.get("headers")
+        self._timeout: ClientTimeout | None = (
+            ClientTimeout(total=timeout) if timeout is not None else client_config.get("timeout")
+        )
         self._base_url = base_url
-        if timeout is not None:
-            self._client_config["timeout"] = ClientTimeout(total=timeout)
+        self._sessions = sessions if sessions is not None else LoopSessions()
+
+    def _use_sessions(self, sessions: LoopSessions) -> None:
+        """Send the requests through ``sessions`` from now on."""
+        self._sessions = sessions
+
+    def _request_url(self, url: str) -> URL:
+        """``url`` resolved against the client's ``base_url``, as an aiohttp session does it.
+
+        Raises:
+            ValueError: If the client's ``base_url`` is not one an aiohttp session takes.
+        """
+        target = URL(url)
+        if self._server_url is None:
+            return target
+        server_url = _session_base_url(self._server_url)
+        return target if target.absolute else server_url.join(target)
+
+    def _request_options(self, options: dict[str, Any]) -> dict[str, Any]:
+        """The client's headers and timeout, with a request's own aiohttp ``options`` over them.
+
+        The request's options win, as they did over the options of a session of the
+        client's own: a header in ``options["headers"]`` replaces the client's header of
+        that name.
+        """
+        headers = CIMultiDict(self._headers or {})
+        headers.update(options.get("headers") or {})
+        defaults = {} if self._timeout is None else {"timeout": self._timeout}
+        return {**defaults, **options, "headers": headers}
 
     def _log_request(self, url: str, method: str) -> None:
         sdk_logger.debug(f"Sending HTTP request: {method} {url}")
@@ -100,13 +186,14 @@ def _prepare_json(
     async def get(self, url: str, model: type[TModel], **kwargs: Any) -> TModel:
         """Send a GET request and parse the JSON response into `model`."""
         url = f"{self._base_url}{url}"
-        async with aiohttp.ClientSession(**self._client_config) as client:
-            self._log_request(url, "GET")
-            async with client.get(url, **kwargs) as response:
-                await handle_api_error(response)
-                self._log_response(url, "GET", response.status)
-                data = await response.json()
-                return parse_obj_as(model, data)
+        target = self._request_url(url)
+        client = await self._sessions.current()
+        self._log_request(url, "GET")
+        async with client.get(target, **self._request_options(kwargs)) as response:
+            await handle_api_error(response)
+            self._log_response(url, "GET", response.status)
+            data = await response.json()
+            return parse_obj_as(model, data)
 
     @handle_client_error
     async def post(
@@ -118,13 +205,16 @@ async def post(
     ) -> TModel:
         """Send a POST request with a JSON body and parse the JSON response into `model`."""
         url = f"{self._base_url}{url}"
-        async with aiohttp.ClientSession(**self._client_config) as client:
-            self._log_request(url, "POST")
-            async with client.post(url, json=self._prepare_json(json), **kwargs) as response:
-                await handle_api_error(response)
-                self._log_response(url, "POST", response.status)
-                data = await response.json()
-                return parse_obj_as(model, data)
+        target = self._request_url(url)
+        client = await self._sessions.current()
+        self._log_request(url, "POST")
+        async with client.post(
+            target, json=self._prepare_json(json), **self._request_options(kwargs)
+        ) as response:
+            await handle_api_error(response)
+            self._log_response(url, "POST", response.status)
+            data = await response.json()
+            return parse_obj_as(model, data)
 
     @handle_client_error
     async def put(
@@ -136,13 +226,16 @@ async def put(
     ) -> TModel:
         """Send a PUT request with a JSON body and parse the JSON response into `model`."""
         url = f"{self._base_url}{url}"
-        async with aiohttp.ClientSession(**self._client_config) as client:
-            self._log_request(url, "PUT")
-            async with client.put(url, json=self._prepare_json(json), **kwargs) as response:
-                await handle_api_error(response)
-                self._log_response(url, "PUT", response.status)
-                data = await response.json()
-                return parse_obj_as(model, data)
+        target = self._request_url(url)
+        client = await self._sessions.current()
+        self._log_request(url, "PUT")
+        async with client.put(
+            target, json=self._prepare_json(json), **self._request_options(kwargs)
+        ) as response:
+            await handle_api_error(response)
+            self._log_response(url, "PUT", response.status)
+            data = await response.json()
+            return parse_obj_as(model, data)
 
     @handle_client_error
     async def patch(
@@ -154,13 +247,16 @@ async def patch(
     ) -> TModel:
         """Send a PATCH request with a JSON body and parse the JSON response into `model`."""
         url = f"{self._base_url}{url}"
-        async with aiohttp.ClientSession(**self._client_config) as client:
-            self._log_request(url, "PATCH")
-            async with client.patch(url, json=self._prepare_json(json), **kwargs) as response:
-                await handle_api_error(response)
-                self._log_response(url, "PATCH", response.status)
-                data = await response.json()
-                return parse_obj_as(model, data)
+        target = self._request_url(url)
+        client = await self._sessions.current()
+        self._log_request(url, "PATCH")
+        async with client.patch(
+            target, json=self._prepare_json(json), **self._request_options(kwargs)
+        ) as response:
+            await handle_api_error(response)
+            self._log_response(url, "PATCH", response.status)
+            data = await response.json()
+            return parse_obj_as(model, data)
 
     @overload
     async def delete(
@@ -190,15 +286,18 @@ async def delete(
     ) -> TModel | None:
         """Send a DELETE request; parse the JSON response into `model` if one is given."""
         url = f"{self._base_url}{url}"
-        async with aiohttp.ClientSession(**self._client_config) as client:
-            self._log_request(url, "DELETE")
-            async with client.delete(url, json=self._prepare_json(json), **kwargs) as response:
-                await handle_api_error(response)
-                self._log_response(url, "DELETE", response.status)
-                if model is None:
-                    return None
-                data = await response.json()
-                return parse_obj_as(model, data)
+        target = self._request_url(url)
+        client = await self._sessions.current()
+        self._log_request(url, "DELETE")
+        async with client.delete(
+            target, json=self._prepare_json(json), **self._request_options(kwargs)
+        ) as response:
+            await handle_api_error(response)
+            self._log_response(url, "DELETE", response.status)
+            if model is None:
+                return None
+            data = await response.json()
+            return parse_obj_as(model, data)
 
 
 class BasePermitApi:
@@ -211,10 +310,21 @@ def __init__(self, config: PermitConfig) -> None:
             config: The Permit SDK configuration.
         """
         self.config = config
+        self._sessions = LoopSessions()
         self.__api_keys = self._build_http_client("/v2/api-key")
 
+    def _use_sessions(self, sessions: LoopSessions) -> None:
+        """Send the requests of this API and of the APIs and clients it holds through ``sessions``.
+
+        A Permit client calls it so that all of its APIs share one session per event loop.
+        """
+        self._sessions = sessions
+        for value in vars(self).values():
+            if isinstance(value, (BasePermitApi, SimpleHttpClient)):
+                value._use_sessions(sessions)  # noqa: SLF001 - SDK-internal
+
     def _build_http_client(
-        self, endpoint_url: str = "", *, use_pdp: bool = False, **kwargs: Any
+        self, endpoint_url: str = "", *, use_pdp: bool = False
     ) -> SimpleHttpClient:
         optional_headers = {}
         if self.config.proxy_facts_via_pdp:
@@ -231,12 +341,11 @@ def _build_http_client(
                 **optional_headers,
             },
         )
-        client_config_dict = client_config.dict()
-        client_config_dict.update(kwargs)
         return SimpleHttpClient(
-            client_config_dict,
+            client_config.dict(),
             base_url=endpoint_url,
             timeout=self.config.api_timeout,
+            sessions=self._sessions,
         )
 
     async def _set_context_from_api_key(self) -> None: