feat: support free-threaded Python builds

.github/workflows/ci.yml

@@ -80,7 +80,9 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.11", "3.12", "3.13", "3.14", "3.15"]
+        python-version: ["3.11", "3.12", "3.13", "3.14", "3.15", "3.14t", "3.15t"]
+    env:
+      UV_PYTHON: ${{ matrix.python-version }}
 
     steps:
       - uses: actions/checkout@v7

CONTRIBUTING.md

@@ -26,7 +26,7 @@ uv run ty check
 uv run pytest
 ```
 
-CI also runs the tests on Python 3.11 to 3.15, and against the oldest FastAPI release the project supports (0.123.0).
+CI also runs the tests on Python 3.11 to 3.15, on the free-threaded 3.14t and 3.15t builds, and against the oldest FastAPI release the project supports (0.123.0).
 
 ## Tests
 

benchmarks/test_perf.py

@@ -221,7 +221,7 @@ def test_jwt_es256_jwks_validate(benchmark):
 
 
 def test_introspection_cache_hit(benchmark):
-    """Opaque token already introspected: the hash and LRU lookup every request pays."""
+    """Opaque token already introspected: the hash and cache lookup every request pays."""
 
     def provider(request: httpx.Request) -> httpx.Response:
         return httpx.Response(

docs/security-model.md

@@ -35,6 +35,7 @@ What the library takes care of, and what stays with you. To report a vulnerabili
 - **ID tokens are untrusted until validated** with `oauth_validate_id_token`.
 - **Introspection answers are cached.** A token revoked at the provider keeps working until its cached answer expires, `cache_ttl` seconds at most.
 - **The discovery URL is configuration.** Never build it from request input.
+- **Sources and validators can be shared across threads.** One instance can serve several threads, each with its own event loop, as on a free-threaded Python build. Requests on one loop share a provider fetch in flight; another loop makes its own. Your validators and `replay_check` run concurrently too, so keep them thread-safe.
 
 ## Status codes
 

pyproject.toml

@@ -22,6 +22,7 @@ classifiers = [
     "Programming Language :: Python :: 3.13",
     "Programming Language :: Python :: 3.14",
     "Programming Language :: Python :: 3.15",
+    "Programming Language :: Python :: Free Threading :: 3 - Stable",
     "Topic :: Internet :: WWW/HTTP :: Session",
     "Topic :: Security",
     "Topic :: Software Development :: Libraries :: Python Modules",
@@ -30,7 +31,6 @@ classifiers = [
 ]
 dependencies = [
     "anyio>=3.7",
-    "async-lru>=2.0",
     "fastapi>=0.123.0,!=0.137.0,!=0.137.1",
     "httpx>=0.25.0",
     "itsdangerous>=2.1",

src/fastapi_multiauth/abc.py

@@ -208,10 +208,13 @@ def _carry_scheme(dependency: Any, scheme: SecurityBase) -> None:
     cls = type(dependency)
     promoted = _SCHEME_CLASSES.get(cls)
     if promoted is None:
-        promoted = _SCHEME_CLASSES[cls] = type(
-            cls.__name__,
-            (cls, SecurityBase),
-            {"__module__": cls.__module__, "__doc__": cls.__doc__},
+        promoted = _SCHEME_CLASSES.setdefault(
+            cls,
+            type(
+                cls.__name__,
+                (cls, SecurityBase),
+                {"__module__": cls.__module__, "__doc__": cls.__doc__},
+            ),
         )
     dependency.__class__ = promoted
     dependency.model = scheme.model

src/fastapi_multiauth/introspection.py

@@ -4,18 +4,18 @@
 import copy
 import logging
 import time
-from collections import OrderedDict
 from collections.abc import Callable, Sequence
 from typing import Any, ClassVar
 from urllib.parse import quote_plus
 
-import anyio
 import httpx
 
 from ._http import DEFAULT_TIMEOUT, _request_json, _require_https
 from .exceptions import UnauthorizedError
 from .jwt import AnyAudience
 from .utils import (
+    SingleFlight,
+    TTLCache,
     ensure_async,
     finite_number,
     hash_token,
@@ -53,15 +53,6 @@ def copy(self) -> dict[str, Any]:
         return copied
 
 
-class _Lookup:
-    """One introspection in flight, awaited by every request carrying its token."""
-
-    def __init__(self) -> None:
-        self.done = anyio.Event()
-        self.answer: _Answer | None = None
-        self.error: Exception | None = None
-
-
 def _basic_authorization(client_id: str, client_secret: str) -> str:
     """`client_secret_basic`: both halves form-encoded before base64 (RFC 6749 §2.3.1)."""
     pair = f"{quote_plus(client_id, safe='')}:{quote_plus(client_secret, safe='')}"
@@ -153,9 +144,8 @@ def __init__(
             ensure_async(claims_to_identity) if claims_to_identity else None
         )
         self._cache_ttl = cache_ttl
-        self._cache_size = cache_size
-        self._cache: OrderedDict[str, tuple[float, _Answer]] = OrderedDict()
-        self._lookups: dict[str, _Lookup] = {}
+        self._cache = TTLCache(cache_size)
+        self._lookups = SingleFlight()
         self._timeout = timeout
         self._client = client
 
@@ -168,7 +158,9 @@ async def __call__(self, token: str, scopes: list[str] | None = None) -> Any:
             HTTPException: 503 if the introspection endpoint cannot be used.
         """
         key = hash_token(token)
-        answer = self._cached(key) or await self._lookup(token, key)
+        answer = self._cache.get(key) or await self._lookups.run(
+            key, lambda: self._lookup(token, key)
+        )
         claims = answer.copy()
         if scopes:
             require_scopes(claims, self._scopes_claim, scopes)
@@ -178,26 +170,11 @@ async def __call__(self, token: str, scopes: list[str] | None = None) -> Any:
         return claims
 
     async def _lookup(self, token: str, key: str) -> _Answer:
-        """The answer for `token`, from the introspection in flight or a new one."""
-        while (lookup := self._lookups.get(key)) is not None:
-            await lookup.done.wait()
-            if lookup.error is not None:
-                raise copy.copy(lookup.error)
-            if lookup.answer is not None:
-                return lookup.answer
-        lookup = self._lookups[key] = _Lookup()
-        try:
-            claims = await self._introspect(token)
-            self._check_claims(claims)
-            answer = _Answer(claims)
-            self._store(key, answer)
-            lookup.answer = answer
-        except Exception as exc:
-            lookup.error = exc
-            raise
-        finally:
-            del self._lookups[key]
-            lookup.done.set()
+        """Introspect `token`, check its claims and cache the answer."""
+        claims = await self._introspect(token)
+        self._check_claims(claims)
+        answer = _Answer(claims)
+        self._store(key, answer)
         return answer
 
     async def _introspect(self, token: str) -> dict[str, Any]:
@@ -258,27 +235,11 @@ def _check_claims(self, claims: dict[str, Any]) -> None:
         if self._issuer is not None and claims.get("iss") != self._issuer:
             raise refusal(UnauthorizedError(), "invalid_issuer")
 
-    def _cached(self, key: str) -> _Answer | None:
-        """A cached answer still within its lifetime, `None` otherwise."""
-        entry = self._cache.get(key)
-        if entry is None:
-            return None
-        expires_at, answer = entry
-        if time.monotonic() >= expires_at:
-            del self._cache[key]
-            return None
-        self._cache.move_to_end(key)
-        return answer
-
     def _store(self, key: str, answer: _Answer) -> None:
         """Cache an active answer for `cache_ttl`, never past the token's `exp`."""
         ttl = self._cache_ttl
         exp = finite_number(answer.claims.get("exp"))
         if exp is not None:
             ttl = min(ttl, exp - time.time())
-        if ttl <= 0:
-            return
-        self._cache[key] = (time.monotonic() + ttl, answer)
-        self._cache.move_to_end(key)
-        while len(self._cache) > self._cache_size:
-            self._cache.popitem(last=False)
+        if ttl > 0:
+            self._cache.put(key, answer, ttl)

src/fastapi_multiauth/jwt.py

@@ -2,16 +2,18 @@
 
 import enum
 import logging
+import threading
 import time
 from collections.abc import Callable, Sequence
 from typing import Any, ClassVar
 
-import anyio
+from anyio.lowlevel import current_token
 
 from ._http import DEFAULT_TIMEOUT, _get_json, _require_https
 from ._imports import require_extra
 from .exceptions import UnauthorizedError
 from .utils import (
+    SingleFlight,
     ensure_async,
     record_verified_jkt,
     refusal,
@@ -118,9 +120,10 @@ def __init__(
         self._refresh_cooldown = refresh_cooldown
         self._timeout = timeout
         self._keys: dict[str, Any] = {}
-        self._fetched_at = 0.0
-        self._attempted_at = float("-inf")
-        self._refresh_lock = anyio.Lock()
+        self._fetched_at = float("-inf")
+        self._attempts: dict[Any, float] = {}
+        self._lock = threading.Lock()
+        self._refreshes = SingleFlight()
 
     async def key_for_token(self, token: str) -> Any:
         """The key that signed `token`, found by its header `kid`.
@@ -159,27 +162,42 @@ def _cache_stale(self) -> bool:
     async def _refresh(self, force: bool = False) -> None:
         """Refetch the JWKS, coalescing concurrent and repeated refreshes.
 
-        At most one attempt per `refresh_cooldown`; a failed fetch falls
+        Requests on one event loop share its refresh in flight. Each loop
+        attempts at most one fetch per `refresh_cooldown`, and none within
+        it of a fetch that succeeded on any loop; a failed fetch falls
         back on cached keys, or raises when the cache is empty.
         """
-        async with self._refresh_lock:
-            # Re-check after acquiring: another request may just have tried.
-            if time.monotonic() - self._attempted_at < self._refresh_cooldown:
-                if not self._keys:
-                    raise _JWKSUnavailable()
-                return
-            if not force and not self._cache_stale():
-                return
-            self._attempted_at = time.monotonic()
-            try:
-                await self._fetch()
-            except _JWKSFetchError as exc:
-                logger.warning("JWKS refresh failed: %s", exc)
-                if not self._keys:
-                    raise _JWKSUnavailable() from exc
-                # Stale fallback: keep validating with the cached keys.
+        loop = current_token()
+        await self._refreshes.run(None, lambda: self._attempt(loop, force))
+
+    async def _attempt(self, loop: Any, force: bool) -> None:
+        if not self._claim_attempt(loop, force):
+            if not self._keys:
+                raise _JWKSUnavailable()
+            return
+        try:
+            await self._fetch()
+        except _JWKSFetchError as exc:
+            logger.warning("JWKS refresh failed: %s", exc)
+            if not self._keys:
+                raise _JWKSUnavailable() from exc
+
+    def _claim_attempt(self, loop: Any, force: bool) -> bool:
+        """Whether `loop` may fetch now, recording the attempt when it may."""
+        now = time.monotonic()
+        cooldown = self._refresh_cooldown
+        with self._lock:
+            last = max(self._fetched_at, self._attempts.get(loop, float("-inf")))
+            if now - last < cooldown or not (force or self._cache_stale()):
+                return False
+            self._attempts = {
+                seen: at for seen, at in self._attempts.items() if now - at < cooldown
+            }
+            self._attempts[loop] = now
+            return True
 
     async def _fetch(self) -> None:
+        started = time.monotonic()
         document = await _get_json(
             self._url,
             timeout=self._timeout,
@@ -221,8 +239,10 @@ async def _fetch(self) -> None:
                 f"JWKS document has no usable signing key "
                 f"({len(document['keys'])} entries)"
             )
-        self._keys = keys
-        self._fetched_at = time.monotonic()
+        with self._lock:
+            if started >= self._fetched_at:
+                self._keys = keys
+                self._fetched_at = started
 
 
 class JWTValidator:

src/fastapi_multiauth/oauth/client.py

@@ -6,11 +6,11 @@
 from urllib.parse import urlsplit
 
 import httpx
-from async_lru import alru_cache
 
 from .._http import DEFAULT_TIMEOUT, _get_json, _require_https
 from .._imports import require_extra
 from ..exceptions import OAuthDiscoveryError, OAuthExchangeError, OAuthUserinfoError
+from ..utils import SingleFlight, TTLCache
 
 __all__ = [
     "OIDCEndpoints",
@@ -21,6 +21,7 @@
 
 _DISCOVERY_SUFFIX = "/.well-known/openid-configuration"
 _DISCOVERY_CACHE_TTL = 3600
+_DISCOVERY_CACHE_SIZE = 32
 
 TokenEndpointAuthMethod = Literal["client_secret_post", "client_secret_basic"]
 
@@ -114,7 +115,15 @@ def get_httpx_client(self) -> Any:
     return _TimeoutOAuth2
 
 
-@alru_cache(maxsize=32, ttl=_DISCOVERY_CACHE_TTL)
+_discovery_cache = TTLCache(_DISCOVERY_CACHE_SIZE)
+_discovery_flights = SingleFlight()
+
+
+def _clear_discovery_cache() -> None:
+    """Forget every cached discovery document."""
+    _discovery_cache.clear()
+
+
 async def oauth_resolve_provider_urls(
     discovery_url: str,
     timeout: float = DEFAULT_TIMEOUT,
@@ -138,6 +147,24 @@ async def oauth_resolve_provider_urls(
             ones only pass from a loopback discovery URL), or names another
             issuer.
     """
+    key = (discovery_url, timeout)
+
+    async def fetch() -> OIDCEndpoints:
+        endpoints = await _fetch_provider_urls(discovery_url, timeout)
+        _discovery_cache.put(key, endpoints, _DISCOVERY_CACHE_TTL)
+        return endpoints
+
+    cached = _discovery_cache.get(key)
+    if cached is not None:
+        return cached
+    return await _discovery_flights.run(key, fetch)
+
+
+setattr(oauth_resolve_provider_urls, "cache_clear", _clear_discovery_cache)  # noqa: B010
+
+
+async def _fetch_provider_urls(discovery_url: str, timeout: float) -> OIDCEndpoints:
+    """Fetch and check one discovery document."""
     _require_https(discovery_url, "OIDC discovery URL")
     cfg = await _get_json(
         discovery_url,