feat(supabase): add with_token() for per-request clients

sdk-compliance.yaml

@@ -408,7 +408,9 @@ features:
   client.authentication_integration.third_party_auth: not_implemented
   client.authentication_integration.cross_client_token_sync:
     status: partially_implemented
-    note: "Every auth event that carries a session sets the Authorization header used by auth, postgrest, storage and functions to that session's access token, and an event without a session resets it to the project key. AsyncClient also passes the token to realtime. The sync Client has no realtime client."
+    note: "Every auth event that carries a session sets the Authorization header used by auth, postgrest, storage and functions to that session's access token, and an event without a session resets it to the project key. AsyncClient also passes the token to realtime. The sync Client has no realtime client. SupabaseClient.with_token returns a client that shares the HTTP session and options and sends a given user JWT to every sub-client, for servers that act for one user per request. Like a supabase-js client with accessToken, it has no auth client: its auth raises."
+    symbols:
+      - SupabaseClient.with_token
   client.authentication_integration.oauth_flow_type:
     status: implemented
     note: "flow_type of AuthClientOptions, passed as the auth argument of the supabase client, and of the auth client constructor."

src/supabase/README.md

@@ -119,6 +119,27 @@ supabase = create_client(
 supabase.table("countries").select("*").retry(False).execute()
 ```
 
+### One client per request on a server
+
+A server such as a FastAPI or Django app acts for the user who sent the request, so that row level security applies to them. Create one client when the app starts and call `with_token()` with the user's JWT in each request. It returns a client that shares the HTTP connections, headers and options of the first one, and only differs in the `Authorization` header. It does no I/O, so creating one per request is cheap. It never refreshes or replaces the token, and leaving it with `async with` does not close the shared connections. It has no auth client, so using its `auth` raises `SupabaseException`.
+
+```python
+import os
+
+from supabase.httpx import create_async_client
+
+shared = create_async_client(os.environ["SUPABASE_URL"], os.environ["SUPABASE_KEY"])
+
+
+async def list_todos(access_token: str) -> object:
+    response = (
+        await shared.with_token(access_token).table("todos").select("*").execute()
+    )
+    return response.data
+```
+
+In FastAPI, read the token with the `HTTPBearer` security dependency, which parses the `Authorization` header and rejects a request without one, and create `shared` with `async with` in the lifespan of the app, so that it is closed when the app shuts down. To check the token, use `shared.auth.get_claims(access_token)`, which caches the signing keys of the project between requests.
+
 Use the supabase client to interface with your database.
 
 ### Sign-up
@@ -299,7 +320,8 @@ SupabaseError
 ├── FunctionsError                  Root of the functions errors.
 ├── RealtimeError                   Root of the realtime errors: NotConnectedError,
 │                                   AuthorizationError, MaxRetriesExceeded
-└── SupabaseException               The client was created with an invalid URL or key.
+└── SupabaseException               The client was created with an invalid URL or key, or `auth`
+                                    was used on a `with_token()` client.
 ```
 
 The response errors inherit both from the root of their package and from `SupabaseAPIError`, so `except AuthApiError` and `except AuthError` keep working. Arguments the client rejects before sending a request raise the built-in `ValueError` or `TypeError`, as they do everywhere in Python, except for a vector batch of the wrong size, which raises `VectorBucketException`. A lost realtime connection is reported as the `websockets.ConnectionClosed` or built-in `TimeoutError` that failed it, not as a `SupabaseError`.

src/supabase/src/supabase/aiohttp.py

@@ -40,7 +40,9 @@ def create_async_client(
     client = AsyncClient(
         supabase_url,
         supabase_key,
-        AsyncAiohttpSession(client=http_client or ClientSession()),
+        AsyncAiohttpSession(
+            client=http_client or ClientSession(), owns_client=http_client is None
+        ),
         headers=headers,
         request_timeout=request_timeout,
         retry=retry,

src/supabase/src/supabase/client.py

@@ -87,6 +87,7 @@
     from supabase_auth.client import SupabaseAuthHttpClient
     from supabase_auth.types import AuthStateChange
     from supabase_functions.client import FunctionsClient
+    from supabase_utils.http.request import Request, Response
     from supabase_utils.types import JSON
 
     from .types import _ServiceClientOptions
@@ -139,7 +140,8 @@
 
 
 class SupabaseException(SupabaseError):
-    """Raised when the client is created with a missing or invalid URL or key."""
+    """Raised when the client is created with a missing or invalid URL or key, or
+    when `auth` is used on a client returned by `with_token()`."""
 
 
 def _find_authorization(headers: Mapping[str, str]) -> str | None:
@@ -153,6 +155,48 @@ def _is_new_api_key(key: str) -> bool:
     return key.startswith("sb_")
 
 
+class _SharedHttpSession:
+    """Sends the requests with a session that another client closes."""
+
+    def __init__(self, session: HttpSession) -> None:
+        self.session = session
+
+    def send(self, request: Request) -> Response:
+        return self.session.send(request)
+
+    def __enter__(self) -> Self:
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc: BaseException | None,
+        tb: TracebackType | None,
+    ) -> None:
+        pass
+
+
+class _SharedAsyncHttpSession:
+    """Sends the requests with a session that another client closes."""
+
+    def __init__(self, session: AsyncHttpSession) -> None:
+        self.session = session
+
+    async def send(self, request: Request) -> Response:
+        return await self.session.send(request)
+
+    async def __aenter__(self) -> Self:
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc: BaseException | None,
+        tb: TracebackType | None,
+    ) -> None:
+        pass
+
+
 class SupabaseClient[
     HttpIO: (SyncHttpIO, AsyncHttpIO),
     Storage: (AsyncSupportedStorage, SyncSupportedStorage),
@@ -267,6 +311,39 @@ def headers(self) -> Mapping[str, str]:
         """
         return MappingProxyType(self._headers)
 
+    def with_token(self, access_token: str) -> Self:
+        """Return a client that makes its requests as the user of `access_token`.
+
+        Use it on a server to act for the user who sent a request, so row
+        level security applies to their JWT. Pass the JWT without the
+        `Bearer ` prefix.
+
+        The returned client shares the HTTP session, the headers and the
+        options of this client, and does no I/O, so creating one per request
+        is cheap. Leaving it or its services with `with` or `async with` does
+        not close the shared HTTP session.
+
+        `access_token` is sent as the `Authorization` header and is never
+        refreshed or replaced. The returned client has no auth client, so
+        using its `auth` raises `SupabaseException`. Use the `auth` of this
+        client instead, for example `auth.get_claims(access_token)` to verify
+        the JWT.
+        """
+        if not access_token:
+            raise ValueError("access_token is required")
+        return self._without_auth(
+            {**self._headers, "Authorization": self._create_auth_header(access_token)}
+        )
+
+    @abstractmethod
+    def _without_auth(self, headers: Mapping[str, str]) -> Self: ...
+
+    def _no_auth_error(self) -> SupabaseException:
+        return SupabaseException(
+            "The client was created with with_token(), so it has no auth"
+            " client. Use the auth of the client that with_token() was called on."
+        )
+
     @property
     @abstractmethod
     def auth(self) -> SupabaseAuthHttpClient[HttpIO, Any]: ...
@@ -469,7 +546,19 @@ def __init__(
         self.realtime_url = self.supabase_url.joinpath("realtime", "v1").with_scheme(
             "wss" if self.supabase_url.scheme == "https" else "ws"
         )
-        self.auth.on_auth_state_change(self._listen_to_auth_events)
+        auth_client = AsyncSupabaseAuthClient(
+            url=str(self.auth_url),
+            http_session=self.http_session,
+            headers=self._headers,
+            auto_refresh_token=self._auth_options.auto_refresh_token,
+            storage=self._auth_options.session_storage,
+            storage_key=self._auth_options.storage_key,
+            flow_type=self._auth_options.flow_type,
+            request_timeout=self._timeout_for(self._auth_options),
+            retry=self._retry_for(self._auth_options),
+        )
+        self._auth: AsyncSupabaseAuthClient | None = auth_client
+        auth_client.on_auth_state_change(self._listen_to_auth_events)
 
     @classmethod
     async def create(
@@ -520,7 +609,8 @@ async def aclose(self) -> None:
         connections opened from `realtime`, which close when their
         `async with` block ends. Closing the client more than once is safe.
         """
-        self.auth.session_manager.stop_auto_refresh()
+        if self._auth is not None:
+            self._auth.session_manager.stop_auto_refresh()
         if self._owns_http_session:
             await self.http_session.__aexit__(None, None, None)
 
@@ -535,20 +625,30 @@ async def __aexit__(
     ) -> None:
         await self.aclose()
 
-    @cached_property
     @override
-    def auth(self) -> AsyncSupabaseAuthClient:
-        return AsyncSupabaseAuthClient(
-            url=str(self.auth_url),
-            http_session=self.http_session,
-            headers=self._headers,
-            auto_refresh_token=self._auth_options.auto_refresh_token,
-            storage=self._auth_options.session_storage,
-            storage_key=self._auth_options.storage_key,
-            flow_type=self._auth_options.flow_type,
-            request_timeout=self._timeout_for(self._auth_options),
-            retry=self._retry_for(self._auth_options),
+    def _without_auth(self, headers: Mapping[str, str]) -> Self:
+        client = type(self)(
+            str(self.supabase_url),
+            self.supabase_key,
+            _SharedAsyncHttpSession(self.http_session),
+            headers=headers,
+            request_timeout=self._request_timeout,
+            retry=self._retry,
+            postgrest=self._postgrest_options,
+            auth=self._auth_options,
+            storage=self._storage_options,
+            functions=self._functions_options,
+            realtime=self._realtime_options,
         )
+        client._auth = None
+        return client
+
+    @property
+    @override
+    def auth(self) -> AsyncSupabaseAuthClient:
+        if self._auth is None:
+            raise self._no_auth_error()
+        return self._auth
 
     @cached_property
     @override
@@ -640,7 +740,19 @@ def __init__(
             storage=storage,
             functions=functions,
         )
-        self.auth.on_auth_state_change(self._listen_to_auth_events)
+        auth_client = SyncSupabaseAuthClient(
+            url=str(self.auth_url),
+            http_session=self.http_session,
+            headers=self._headers,
+            auto_refresh_token=self._auth_options.auto_refresh_token,
+            storage=self._auth_options.session_storage,
+            storage_key=self._auth_options.storage_key,
+            flow_type=self._auth_options.flow_type,
+            request_timeout=self._timeout_for(self._auth_options),
+            retry=self._retry_for(self._auth_options),
+        )
+        self._auth: SyncSupabaseAuthClient | None = auth_client
+        auth_client.on_auth_state_change(self._listen_to_auth_events)
 
     @classmethod
     def create(
@@ -688,7 +800,8 @@ def close(self) -> None:
         A session passed in by the caller is left open. Closing the client
         more than once is safe.
         """
-        self.auth.session_manager.stop_auto_refresh()
+        if self._auth is not None:
+            self._auth.session_manager.stop_auto_refresh()
         if self._owns_http_session:
             self.http_session.__exit__(None, None, None)
 
@@ -703,20 +816,29 @@ def __exit__(
     ) -> None:
         self.close()
 
-    @cached_property
     @override
-    def auth(self) -> SyncSupabaseAuthClient:
-        return SyncSupabaseAuthClient(
-            url=str(self.auth_url),
-            http_session=self.http_session,
-            headers=self._headers,
-            auto_refresh_token=self._auth_options.auto_refresh_token,
-            storage=self._auth_options.session_storage,
-            storage_key=self._auth_options.storage_key,
-            flow_type=self._auth_options.flow_type,
-            request_timeout=self._timeout_for(self._auth_options),
-            retry=self._retry_for(self._auth_options),
+    def _without_auth(self, headers: Mapping[str, str]) -> Self:
+        client = type(self)(
+            str(self.supabase_url),
+            self.supabase_key,
+            _SharedHttpSession(self.http_session),
+            headers=headers,
+            request_timeout=self._request_timeout,
+            retry=self._retry,
+            postgrest=self._postgrest_options,
+            auth=self._auth_options,
+            storage=self._storage_options,
+            functions=self._functions_options,
         )
+        client._auth = None
+        return client
+
+    @property
+    @override
+    def auth(self) -> SyncSupabaseAuthClient:
+        if self._auth is None:
+            raise self._no_auth_error()
+        return self._auth
 
     @cached_property
     @override

src/supabase/src/supabase/httpx.py

@@ -45,7 +45,7 @@ def create_client(
     client = Client(
         supabase_url,
         supabase_key,
-        HttpxSession(client=httpx_client),
+        HttpxSession(client=httpx_client, owns_client=http_client is None),
         headers=headers,
         request_timeout=request_timeout,
         retry=retry,
@@ -87,7 +87,7 @@ def create_async_client(
     client = AsyncClient(
         supabase_url,
         supabase_key,
-        AsyncHttpxSession(client=httpx_client),
+        AsyncHttpxSession(client=httpx_client, owns_client=http_client is None),
         headers=headers,
         request_timeout=request_timeout,
         retry=retry,

src/supabase/tests/test_client.py

@@ -1,3 +1,4 @@
+from concurrent.futures import ThreadPoolExecutor
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import httpx
@@ -524,3 +525,13 @@ def test_factories_reject_an_invalid_request_timeout_before_creating_a_session(
         aiohttp_session.assert_not_called()
         sync_httpx_client.assert_not_called()
         async_httpx_client.assert_not_called()
+
+
+def test_sync_client_shares_one_auth_client_between_threads() -> None:
+    with SyncHttpxClient() as http_client:
+        client = Client(URL, KEY, HttpxSession(client=http_client))
+        with ThreadPoolExecutor(max_workers=8) as pool:
+            futures = [pool.submit(lambda: client.auth) for _ in range(32)]
+            auth_clients = [future.result() for future in futures]
+
+    assert all(auth is auth_clients[0] for auth in auth_clients)