httpx OAuth2 client-credentials client with cached tokens and Link pagination: working versions (tested 2026-10-15)

httpx==0.28.1
respx==0.23.1
pytest==9.1.1

import time
import httpx

class OAuthApiClient:
    def __init__(self, base_url: str, token_url: str, client_id: str, client_secret: str,
                 clock=time.monotonic, leeway: float = 30):
        self.token_url = token_url
        self.auth = (client_id, client_secret)
        self.clock = clock
        self.leeway = leeway  # refresh a little early so a token never expires mid-request
        self.token: str | None = None
        self.expires_at = 0.0
        self.http = httpx.Client(base_url=base_url, timeout=10)

    def _fetch_token(self) -> None:
        # client_credentials: credentials go in HTTP Basic auth, grant type in the form body
        r = self.http.post(self.token_url, data={"grant_type": "client_credentials"}, auth=self.auth)
        r.raise_for_status()
        body = r.json()
        self.token = body["access_token"]
        self.expires_at = self.clock() + body.get("expires_in", 3600) - self.leeway

    def _bearer(self) -> str:
        if self.token is None or self.clock() >= self.expires_at:
            self._fetch_token()
        return self.token

    def _get(self, url: str, params: dict | None = None) -> httpx.Response:
        r = self.http.get(url, params=params, headers={"Authorization": f"Bearer {self._bearer()}"})
        if r.status_code == 401:  # token revoked or expired early: refresh once and retry
            self._fetch_token()
            r = self.http.get(url, params=params, headers={"Authorization": f"Bearer {self.token}"})
        r.raise_for_status()
        return r

    def list_items(self, path: str, params: dict | None = None):
        url = path
        while url:
            r = self._get(url, params)
            yield from r.json()
            url = r.links.get("next", {}).get("url")  # httpx parses Link headers by rel
            params = None  # the next link already carries its own query string

    def close(self) -> None:
        self.http.close()

import httpx, respx
from client import OAuthApiClient

API = "https://api.example.com"
TOKEN_URL = "https://auth.example.com/oauth/token"

class FakeClock:
    def __init__(self):
        self.now = 0.0

    def __call__(self):
        return self.now

def make_client(clock=None):
    return OAuthApiClient(API, TOKEN_URL, "id", "secret", clock=clock or FakeClock())

def token_resp(token, expires_in=3600):
    return httpx.Response(200, json={"access_token": token, "expires_in": expires_in})

@respx.mock
def test_token_is_fetched_once_and_cached():
    token = respx.post(TOKEN_URL).mock(side_effect=[token_resp("t1")])
    respx.get(f"{API}/items").mock(return_value=httpx.Response(200, json=[]))
    c = make_client()
    list(c.list_items("/items"))
    list(c.list_items("/items"))
    assert token.call_count == 1
    assert token.calls[0].request.headers["Authorization"].startswith("Basic ")
    c.close()

@respx.mock
def test_refreshes_token_once_on_401():
    token = respx.post(TOKEN_URL).mock(side_effect=[token_resp("old"), token_resp("new")])
    items = respx.get(f"{API}/items").mock(
        side_effect=[httpx.Response(401), httpx.Response(200, json=[1])]
    )
    c = make_client()
    assert list(c.list_items("/items")) == [1]
    assert token.call_count == 2
    assert items.calls[1].request.headers["Authorization"] == "Bearer new"
    c.close()

@respx.mock
def test_refreshes_token_after_expiry_minus_leeway():
    clock = FakeClock()
    token = respx.post(TOKEN_URL).mock(side_effect=[token_resp("a", expires_in=100), token_resp("b")])
    respx.get(f"{API}/items").mock(return_value=httpx.Response(200, json=[]))
    c = make_client(clock)
    list(c.list_items("/items"))
    clock.now = 71  # 100s lifetime minus 30s leeway has passed
    list(c.list_items("/items"))
    assert token.call_count == 2
    c.close()

@respx.mock
def test_follows_link_header_pagination():
    respx.get(f"{API}/items", params={"page": "2"}).mock(
        return_value=httpx.Response(200, json=[3, 4])
    )
    respx.get(f"{API}/items").mock(
        return_value=httpx.Response(
            200, json=[1, 2],
            headers={"Link": f'<{API}/items?page=2>; rel="next"'},
        )
    )
    respx.post(TOKEN_URL).mock(return_value=token_resp("t"))
    c = make_client()
    assert list(c.list_items("/items")) == [1, 2, 3, 4]
    c.close()

[pytest]
pythonpath = .