refactor(sdk): deduplicate shared logic and fix type annotations

- Extract _parse_error_response into oauth/_parsing.py shared by sync and async clients
- Extract _validate_and_enrich and _copy_metadata helpers in discovery to eliminate duplication
- Replace copy.deepcopy with dataclasses.replace for Metadata copying
- Extract _is_token_valid module function in clientcreds/token_source.py
- Extract _handle_poll_error helper in authflow/device.py for polling error classification
- Add in-memory token cache to authflow/token_source.py to avoid redundant store reads
- Fix _with_file_lock type annotation from object to Callable[[], None]
- Remove unused request parameter from Django middleware _ensure_configured

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: appleboy

src/authgate/authflow/device.py

@@ -47,6 +47,23 @@ def run_device_flow(
     return _poll_device_code(client, auth)
 
 
+def _handle_poll_error(exc: OAuthError) -> str:
+    """Classify a device-code polling error.
+
+    Returns ``"continue"`` or ``"slow_down"`` for retriable errors.
+    Raises the appropriate exception for terminal errors.
+    """
+    if exc.code == "authorization_pending":
+        return "continue"
+    if exc.code == "slow_down":
+        return "slow_down"
+    if exc.code == "expired_token":
+        raise TokenExpiredError("device code expired") from exc
+    if exc.code == "access_denied":
+        raise AccessDeniedError("access denied by user") from exc
+    raise AuthFlowError(f"exchange device code: {exc}") from exc
+
+
 def _poll_device_code(client: OAuthClient, auth: DeviceAuth) -> Token:
     """Poll the token endpoint until the user authorizes or the code expires."""
     interval = max(auth.interval, 5)
@@ -61,16 +78,9 @@ def _poll_device_code(client: OAuthClient, auth: DeviceAuth) -> Token:
         try:
             return client.exchange_device_code(auth.device_code)
         except OAuthError as exc:
-            if exc.code == "authorization_pending":
-                continue
-            if exc.code == "slow_down":
+            signal = _handle_poll_error(exc)
+            if signal == "slow_down":
                 interval += 5
-                continue
-            if exc.code == "expired_token":
-                raise TokenExpiredError("device code expired") from exc
-            if exc.code == "access_denied":
-                raise AccessDeniedError("access denied by user") from exc
-            raise AuthFlowError(f"exchange device code: {exc}") from exc
 
 
 async def async_run_device_flow(
@@ -108,13 +118,6 @@ async def _async_poll_device_code(client: AsyncOAuthClient, auth: DeviceAuth) ->
         try:
             return await client.exchange_device_code(auth.device_code)
         except OAuthError as exc:
-            if exc.code == "authorization_pending":
-                continue
-            if exc.code == "slow_down":
+            signal = _handle_poll_error(exc)
+            if signal == "slow_down":
                 interval += 5
-                continue
-            if exc.code == "expired_token":
-                raise TokenExpiredError("device code expired") from exc
-            if exc.code == "access_denied":
-                raise AccessDeniedError("access denied by user") from exc
-            raise AuthFlowError(f"exchange device code: {exc}") from exc

src/authgate/authflow/token_source.py

@@ -46,18 +46,25 @@ def __init__(
         self._client = client
         self._store = store
         self._lock = threading.RLock()
+        self._cached: Token | None = None
         self._inflight: threading.Event | None = None
         self._inflight_result: Token | None = None
         self._inflight_error: Exception | None = None
 
     def token(self) -> Token:
         """Return a valid token, refreshing from store or server as needed."""
-        # Fast path: check for valid cached token
+        # Fast path: check in-memory cache first
+        if self._cached is not None and not self._cached.is_expired():
+            return self._cached
+
+        # Check persistent store
         if self._store is not None:
             try:
                 stored = self._store.load(self._client.client_id)
                 if stored.is_valid():
-                    return _credstore_to_oauth(stored)
+                    tok = _credstore_to_oauth(stored)
+                    self._cached = tok
+                    return tok
             except NotFoundError:
                 pass
 
@@ -100,12 +107,15 @@ def _do_refresh(self) -> Token:
             try:
                 stored = self._store.load(self._client.client_id)
                 if stored.is_valid():
-                    return _credstore_to_oauth(stored)
+                    tok = _credstore_to_oauth(stored)
+                    self._cached = tok
+                    return tok
 
                 # Try refresh if we have a refresh token
                 if stored.refresh_token:
                     refreshed = self._client.refresh_token(stored.refresh_token)
                     self._save_token(refreshed)
+                    self._cached = refreshed
                     return refreshed
             except NotFoundError:
                 pass
@@ -115,6 +125,7 @@ def _do_refresh(self) -> Token:
     def save_token(self, token: Token) -> None:
         """Persist a token to the store (if configured)."""
         with self._lock:
+            self._cached = token
             self._save_token(token)
 
     def _save_token(self, token: Token) -> None:

src/authgate/clientcreds/token_source.py

@@ -13,6 +13,15 @@
 _DEFAULT_EXPIRY_DELTA = 30.0  # seconds
 
 
+def _is_token_valid(token: Token | None, expiry_delta: float) -> bool:
+    """Check whether a token is present and not expired."""
+    if token is None or not token.access_token:
+        return False
+    if token.expires_at == 0:
+        return True
+    return (time.time() + expiry_delta) < token.expires_at
+
+
 class TokenSource:
     """Thread-safe, auto-caching token source for client credentials (sync).
 
@@ -39,23 +48,16 @@ def token(self) -> Token:
         """Return a valid access token, fetching a new one if expired."""
         # Fast path
         with self._lock:
-            if self._token is not None and self._is_valid():
-                return self._token
+            if _is_token_valid(self._token, self._expiry_delta):
+                return self._token  # type: ignore[return-value]
 
         return self._slow_path()
 
-    def _is_valid(self) -> bool:
-        if self._token is None or not self._token.access_token:
-            return False
-        if self._token.expires_at == 0:
-            return True
-        return (time.time() + self._expiry_delta) < self._token.expires_at
-
     def _slow_path(self) -> Token:
         with self._lock:
             # Re-check after acquiring lock
-            if self._token is not None and self._is_valid():
-                return self._token
+            if _is_token_valid(self._token, self._expiry_delta):
+                return self._token  # type: ignore[return-value]
 
             if self._inflight is not None:
                 event = self._inflight
@@ -104,18 +106,11 @@ def __init__(
 
     async def token(self) -> Token:
         """Return a valid access token, fetching a new one if expired."""
-        if self._token is not None and self._is_valid():
-            return self._token
+        if _is_token_valid(self._token, self._expiry_delta):
+            return self._token  # type: ignore[return-value]
 
         async with self._lock:
-            if self._token is not None and self._is_valid():
-                return self._token
+            if _is_token_valid(self._token, self._expiry_delta):
+                return self._token  # type: ignore[return-value]
             self._token = await self._client.client_credentials(self._scopes)
             return self._token
-
-    def _is_valid(self) -> bool:
-        if self._token is None or not self._token.access_token:
-            return False
-        if self._token.expires_at == 0:
-            return True
-        return (time.time() + self._expiry_delta) < self._token.expires_at

src/authgate/credstore/file_store.py

@@ -6,6 +6,7 @@
 import json
 import os
 import time
+from collections.abc import Callable
 from typing import Generic, TypeVar
 
 from authgate.credstore.protocols import Codec
@@ -143,10 +144,10 @@ def _ensure_dir(self) -> None:
         if parent:
             os.makedirs(parent, mode=0o700, exist_ok=True)
 
-    def _with_file_lock(self, fn: object) -> None:
+    def _with_file_lock(self, fn: Callable[[], None]) -> None:
         lock = _FileLock(self._file_path + ".lock")
         lock.acquire()
         try:
-            fn()  # type: ignore[operator]
+            fn()
         finally:
             lock.release()

src/authgate/discovery/async_client.py

@@ -3,18 +3,20 @@
 from __future__ import annotations
 
 import asyncio
-import copy
 import time
 
 import httpx
 
-from authgate.discovery.client import _parse_metadata
+from authgate.discovery.client import (
+    _DEFAULT_CACHE_TTL,
+    _WELL_KNOWN_PATH,
+    _copy_metadata,
+    _parse_metadata,
+    _validate_and_enrich,
+)
 from authgate.discovery.models import Metadata
 from authgate.exceptions import DiscoveryError
 
-_WELL_KNOWN_PATH = "/.well-known/openid-configuration"
-_DEFAULT_CACHE_TTL = 3600.0
-
 
 class AsyncDiscoveryClient:
     """OIDC discovery client with caching (async)."""
@@ -36,13 +38,13 @@ def __init__(
     async def fetch(self) -> Metadata:
         """Retrieve the OIDC provider metadata, using the cache if still valid."""
         if self._cached is not None and (time.time() - self._fetched_at) < self._cache_ttl:
-            return copy.deepcopy(self._cached)
+            return _copy_metadata(self._cached)
         return await self._refresh()
 
     async def _refresh(self) -> Metadata:
         async with self._lock:
             if self._cached is not None and (time.time() - self._fetched_at) < self._cache_ttl:
-                return copy.deepcopy(self._cached)
+                return _copy_metadata(self._cached)
 
             url = self._issuer_url + _WELL_KNOWN_PATH
             resp = await self._http.get(url)
@@ -51,20 +53,8 @@ async def _refresh(self) -> Metadata:
 
             body = resp.json()
             meta = _parse_metadata(body)
-
-            issuer = meta.issuer.rstrip("/")
-            if issuer != self._issuer_url:
-                raise DiscoveryError(
-                    f"discovery: issuer mismatch: got {meta.issuer!r},"
-                    f" expected {self._issuer_url!r}"
-                )
-
-            if not meta.device_authorization_endpoint:
-                meta.device_authorization_endpoint = issuer + "/oauth/device/code"
-
-            if not meta.introspection_endpoint:
-                meta.introspection_endpoint = issuer + "/oauth/introspect"
+            _validate_and_enrich(meta, self._issuer_url)
 
             self._cached = meta
             self._fetched_at = time.time()
-            return copy.deepcopy(meta)
+            return _copy_metadata(meta)

src/authgate/discovery/client.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-import copy
+import dataclasses
 import threading
 import time
 
@@ -15,6 +15,35 @@
 _DEFAULT_CACHE_TTL = 3600.0  # 1 hour
 
 
+def _copy_metadata(meta: Metadata) -> Metadata:
+    """Return a shallow copy of Metadata with independent list fields."""
+    return dataclasses.replace(
+        meta,
+        response_types_supported=list(meta.response_types_supported),
+        subject_types_supported=list(meta.subject_types_supported),
+        id_token_signing_alg_values_supported=list(meta.id_token_signing_alg_values_supported),
+        scopes_supported=list(meta.scopes_supported),
+        token_endpoint_auth_methods_supported=list(meta.token_endpoint_auth_methods_supported),
+        grant_types_supported=list(meta.grant_types_supported),
+        claims_supported=list(meta.claims_supported),
+        code_challenge_methods_supported=list(meta.code_challenge_methods_supported),
+    )
+
+
+def _validate_and_enrich(meta: Metadata, expected_issuer: str) -> Metadata:
+    """Validate issuer and fill in default endpoint URLs."""
+    issuer = meta.issuer.rstrip("/")
+    if issuer != expected_issuer:
+        raise DiscoveryError(
+            f"discovery: issuer mismatch: got {meta.issuer!r}, expected {expected_issuer!r}"
+        )
+    if not meta.device_authorization_endpoint:
+        meta.device_authorization_endpoint = issuer + "/oauth/device/code"
+    if not meta.introspection_endpoint:
+        meta.introspection_endpoint = issuer + "/oauth/introspect"
+    return meta
+
+
 class DiscoveryClient:
     """OIDC discovery client with caching (sync)."""
 
@@ -39,13 +68,13 @@ def fetch(self) -> Metadata:
         """
         with self._lock:
             if self._cached is not None and (time.time() - self._fetched_at) < self._cache_ttl:
-                return copy.deepcopy(self._cached)
+                return _copy_metadata(self._cached)
         return self._refresh()
 
     def _refresh(self) -> Metadata:
         with self._lock:
             if self._cached is not None and (time.time() - self._fetched_at) < self._cache_ttl:
-                return copy.deepcopy(self._cached)
+                return _copy_metadata(self._cached)
 
             url = self._issuer_url + _WELL_KNOWN_PATH
             resp = self._http.get(url)
@@ -54,23 +83,11 @@ def _refresh(self) -> Metadata:
 
             body = resp.json()
             meta = _parse_metadata(body)
-
-            issuer = meta.issuer.rstrip("/")
-            if issuer != self._issuer_url:
-                raise DiscoveryError(
-                    f"discovery: issuer mismatch: got {meta.issuer!r},"
-                    f" expected {self._issuer_url!r}"
-                )
-
-            if not meta.device_authorization_endpoint:
-                meta.device_authorization_endpoint = issuer + "/oauth/device/code"
-
-            if not meta.introspection_endpoint:
-                meta.introspection_endpoint = issuer + "/oauth/introspect"
+            _validate_and_enrich(meta, self._issuer_url)
 
             self._cached = meta
             self._fetched_at = time.time()
-            return copy.deepcopy(meta)
+            return _copy_metadata(meta)
 
 
 def _get_str(body: dict[str, object], key: str) -> str:

src/authgate/middleware/django.py

@@ -40,7 +40,7 @@ def __init__(self, get_response: Callable[[HttpRequest], HttpResponse]) -> None:
         self._mode = ValidationMode.TOKEN_INFO
         self._required_scopes: list[str] = []
 
-    def _ensure_configured(self, request: HttpRequest) -> None:
+    def _ensure_configured(self) -> None:
         if self._client is not None:
             return
         from django.conf import settings
@@ -50,7 +50,7 @@ def _ensure_configured(self, request: HttpRequest) -> None:
         self._required_scopes = getattr(settings, "AUTHGATE_REQUIRED_SCOPES", [])
 
     def __call__(self, request: HttpRequest) -> HttpResponse:
-        self._ensure_configured(request)
+        self._ensure_configured()
 
         if self._client is None:
             return self.get_response(request)

src/authgate/oauth/_parsing.py

@@ -0,0 +1,26 @@
+"""Shared OAuth response parsing utilities."""
+
+from __future__ import annotations
+
+import httpx
+
+from authgate.exceptions import OAuthError
+
+
+def _parse_error_response(resp: httpx.Response) -> OAuthError:
+    """Parse an OAuth error response body."""
+    try:
+        body = resp.json()
+        if isinstance(body, dict) and body.get("error"):
+            return OAuthError(
+                code=body["error"],
+                description=body.get("error_description", ""),
+                status_code=resp.status_code,
+            )
+    except Exception:
+        pass
+    return OAuthError(
+        code=resp.reason_phrase or "server_error",
+        description=resp.text,
+        status_code=resp.status_code,
+    )