Skip to content

Authentication

Two ways to send a Bearer Authorization header — provide at most one:

async with HTTPClient(
    base_url="https://api.example.com/", bearer_token="my-static-token"
) as client:
    await client.get("items/7")

bearer_token is a static string, sent as-is on every request. For a token that expires or rotates, pass bearer_auth instead — an async callable (sync callable on SyncHTTPClient) resolved fresh on every request, not just once when the client is created:

async def get_current_token() -> str:
    return await token_store.get_access_token()  # e.g. refreshed from a cache or auth server


async with HTTPClient(base_url="https://api.example.com/", bearer_auth=get_current_token) as client:
    await client.get("items/7")  # calls get_current_token() for this request
    await client.get("items/8")  # calls it again — always the latest token

For the most common reason to need bearer_auth — an OAuth 2 client-credentials token that expires — lothc ships a ready-made provider, so you don't write that callable yourself; see OAuth 2 client credentials below.

bearer_token/bearer_auth both only ever produce a Bearer Authorization header. For a Basic-auth username/password pair instead, pass basic_auth — provide at most one of the three:

async with HTTPClient(
    base_url="https://api.example.com/", basic_auth=("my-username", "my-password")
) as client:
    await client.get("items/7")

password may be None for a username with no password. There's no other custom-auth-scheme option today.

Every verb also takes skip_auth, which omits the Authorization header for that one call — useful when a client configured with bearer_token/bearer_auth/basic_auth also needs to hit a differently-authenticated target through the same instance, e.g. a presigned S3 URL that must never see your API's own token:

async with HTTPClient(
    base_url="https://api.example.com/", bearer_token="my-static-token"
) as client:
    await client.get("items/7")  # gets the Authorization header
    await client.get("https://presigned-bucket.example.com/file", skip_auth=True)  # doesn't

skip_auth=True also skips calling bearer_auth for that request — if refreshing the token is expensive (a network round-trip to an auth server, say), that cost isn't paid on a skipped call.

OAuth 2 client credentials

OAuthProvider is a bearer_auth provider for the OAuth 2 client-credentials grant (RFC 6749 §4.4). Pass an instance where you'd pass the callable, and every request gets a valid access token, minted or refreshed behind the scenes only when needed:

from lothc import HTTPClient, OAuthProvider

async with HTTPClient(
    base_url="https://api.example.com/",
    bearer_auth=OAuthProvider(
        token_url="https://auth.example.com/oauth/token",
        client_id="my-client-id",
        client_secret="my-client-secret",
    ),
) as client:
    await client.get("items/7")

The defaults do exactly what the RFC says: POST token_url as application/x-www-form-urlencoded with grant_type=client_credentials, the client credentials as an HTTP Basic Authorization header (the RFC's preferred placement, with each half percent-encoded first per §2.3.1, so a secret containing :, %, + or a space survives the trip), and the standard access_token/expires_in/refresh_token JSON fields decoded from the response. Two optional knobs on that request: scope="read:things write:things" adds a scope field, and client_auth="body" puts client_id/client_secret in the form body instead of the header, for servers that only accept them there.

expires_in is only RECOMMENDED by the RFC, and some servers leave it out. Pass default_expires_in=3600.0 to tell lothc how long such a token should be trusted for; without it, a response with no expires_in is an error (OAuthTokenError, see below) rather than a token cached forever.

Token lifecycle

  • The first request mints a token and remembers access_token, refresh_token (if the server sent one) and an absolute expiry, now + expires_in.
  • Every later request is a timestamp compare, no lock, no HTTP — until fewer than refresh_leeway seconds (default 300.0, i.e. five minutes) remain, at which point the token is treated as already expired and renewed. The leeway is clamped to half the token's own lifetime, so a server that hands out 60-second tokens gets a 30-second leeway rather than a token that's "expired" the moment it arrives and re-minted on every call.
  • Renewal prefers a refresh: if the last response carried a refresh_token, it sends grant_type=refresh_token. If it didn't, or the refresh is rejected with a 400 (an invalid_grant because the refresh token itself expired or was revoked), it falls back to minting a fresh token with the client-credentials grant. A refresh response that omits refresh_token keeps the one it was refreshed with, as the RFC allows. Any other failure to obtain a token — a 401/403/429 on the refresh, a 5xx from the token endpoint, a transport error, a decode error, a payload with no access_token — is not a fallback case: it is raised from the request that triggered the renewal as an OAuthTokenError, with the original exception (HTTPResponseError, HTTPTransportError, the decode library's own error, ...) as its __cause__ and the token endpoint in .token_url. See Error handling for why this one is wrapped when nothing else in lothc is.
  • Concurrent requests that all land inside the leeway window share one renewal (a lock); the rest wait for it and reuse its result rather than each hitting the token endpoint.

Each token request goes through its own short-lived client, built by calling client_factory with no arguments (default HTTPClient). Anything the token endpoint needs that the API client doesn't — a shorter timeout, a proxy, a private CA, an mTLS identity — goes through a functools.partial of build:

from functools import partial

OAuthProvider(
    token_url="https://auth.example.com/oauth/token",
    client_id="my-client-id",
    client_secret="my-client-secret",
    client_factory=partial(HTTPClient, timeout=5.0, proxy="http://proxy.internal:3128"),
)

The credentials never need to be part of the factory: the provider builds the Authorization header itself, per request.

Persisting the token across restarts

token_cache_dir adds persistence, and only persistence — it never changes when a token is renewed:

from pathlib import Path

OAuthProvider(
    token_url="https://auth.example.com/oauth/token",
    client_id="my-client-id",
    client_secret="my-client-secret",
    token_cache_dir=Path("~/.cache/my-app").expanduser(),
)

Each provider gets its own file in that directory, named for the token URL's host, the client ID and a short hash of the token URL, client ID and scope, e.g. auth.example.com-my-client-id-825a9a6162ca.json. So several providers (different clients, token URLs or scopes) can share one directory without overwriting each other, and you can still tell the files apart by eye. Characters that aren't safe in a filename are replaced with _ in the readable part; the hash always uses the real values.

After every mint/refresh the token triple is written to that file as JSON, atomically (a temp file in the same directory, then replaced into place) with 0600 permissions, since it holds live credentials. It's meant to be readable by a person checking why a token did or didn't renew, so the expiry is an ISO 8601 UTC timestamp at whole-second precision, not an epoch float:

{
  "token_url": "https://auth.example.com/oauth/token",
  "client_id": "my-client-id",
  "scope": null,
  "access_token": "eyJ...",
  "refresh_token": null,
  "expires_at": "2026-09-08T10:56:40Z",
  "expires_in": 3600
}

On construction the file is read back, so a restarted process picks up where the last one left off: a still-valid token is used as-is, one inside the leeway window is renewed via the cached refresh token (expires_in, the lifetime the token was issued with, is what the leeway clamp above is computed from). The file also records token_url, client_id and scope; a mismatch on any of them reads as "no cache", so one client is never handed another client's token, or a token minted for a different set of scopes. A corrupt or unreadable file is never fatal — it's ignored and overwritten on the next mint. The directory is the one thing checked up front: if it doesn't exist, constructing the provider raises FileNotFoundError immediately rather than failing on the first write.

The file isn't encrypted: like the AWS CLI's ~/.aws/sso/cache and gcloud's credentials, it relies on 0600 permissions. Encrypting it with the client secret would add little, since anyone who can read the secret can mint a fresh token anyway.

Non-RFC token endpoints

Plenty of real APIs hand out bearer tokens without speaking RFC 6749: a JSON body instead of form-encoding, every field renamed. Describe such an endpoint with your own pydantic/msgspec models, and lothc reuses its existing json=/response_data_type machinery to send and decode them. The contract is on the attribute names, never the wire names:

  • token_request must be constructible as Cls(client_id=..., client_secret=...); any other field it declares (grant_type, audience, ...) needs a default.
  • token_response must expose .access_token: str and .expires_in: int (or int | None, with default_expires_in covering the None). .refresh_token is read if present (getattr(..., "refresh_token", None)) but isn't required — an API that never issues one needs no field for it at all.
  • token_refresh_request, optional, must be constructible as Cls(refresh_token=...). Without it, renewal always means minting a fresh token, even if the response carried a refresh_token — lothc has no way to know how this API spells a refresh request.

Provide token_request and token_response together, or neither. client_auth is ignored on this path: the request model carries the credentials, so nothing is sent as HTTP Basic.

from pydantic import BaseModel, ConfigDict, Field


class TokenRequest(BaseModel):
    model_config = ConfigDict(validate_by_name=True, serialize_by_alias=True)

    client_id: str = Field(alias="clientID")
    client_secret: str = Field(alias="clientSecret")
    grant_type: str = Field(default="client_credentials", alias="grantType")


class TokenResponse(BaseModel):
    access_token: str = Field(alias="accessToken")
    expires_in: int = Field(alias="expiresIn")
    # This API never returns a refresh token, so there's no field for it at all — leave
    # `token_refresh_request` out too, and renewal simply mints a fresh token every time.


OAuthProvider(
    token_url="https://auth.example.com/v2/token",
    client_id="my-client-id",
    client_secret="my-client-secret",
    token_request=TokenRequest,
    token_response=TokenResponse,
)

The model_config line on the request model is load-bearing, not boilerplate: lothc constructs the model by attribute name, and validate_by_name=True is what lets an aliased pydantic model accept client_id= at all. serialize_by_alias=True is no longer needed, since lothc encodes every pydantic model by its aliases unless the model explicitly sets serialize_by_alias=False (the same way msgspec's rename=/name= works in both directions); it's harmless to keep.

The msgspec spelling of the same thing needs no config — name= covers both directions:

import msgspec


class TokenRequest(msgspec.Struct):
    client_id: str = msgspec.field(name="clientID")
    client_secret: str = msgspec.field(name="clientSecret")
    grant_type: str = msgspec.field(default="client_credentials", name="grantType")


class TokenResponse(msgspec.Struct):
    access_token: str = msgspec.field(name="accessToken")
    expires_in: int = msgspec.field(name="expiresIn")

If the API does issue refresh tokens, declare refresh_token on the response model (it's read if present, see above) and add a third model for the refresh call:

class TokenResponse(BaseModel):
    access_token: str = Field(alias="accessToken")
    expires_in: int = Field(alias="expiresIn")
    refresh_token: str | None = Field(default=None, alias="refreshToken")


class TokenRefreshRequest(BaseModel):
    model_config = ConfigDict(validate_by_name=True, serialize_by_alias=True)

    refresh_token: str = Field(alias="refreshToken")
    grant_type: str = Field(default="refresh_token", alias="grantType")


OAuthProvider(
    token_url="https://auth.example.com/v2/token",
    client_id="my-client-id",
    client_secret="my-client-secret",
    token_request=TokenRequest,
    token_refresh_request=TokenRefreshRequest,
    token_response=TokenResponse,
)

Revoked tokens

A token the server rejects with a 401 before it expires (revoked, rotated) is discarded automatically. The client calls provider.invalidate() and, for an idempotent verb (GET/PUT/DELETE/HEAD), retries the request once with a freshly obtained token. A POST or PATCH is never replayed, since it may not be safe to repeat: it raises the 401, and the next call gets a fresh token. A second 401 on the retry is raised as-is, so a request that genuinely lacks access fails after one extra token fetch rather than looping.

invalidate() is public too, and takes the rejected token optionally, becoming a no-op if a new one has been obtained since (so two concurrent 401s don't each discard the other's fresh token). The client knows nothing about OAuth here: any bearer_auth callable with an invalidate(stale_access_token=None) method gets the same behaviour.

Sync

SyncOAuthProvider is the mirror for SyncHTTPClient — same constructor, same lifecycle, a plain def __call__, and a client_factory defaulting to SyncHTTPClient (so partial(SyncHTTPClient, timeout=5.0) is the sync spelling of the example above):

from lothc import SyncHTTPClient, SyncOAuthProvider

with SyncHTTPClient(
    base_url="https://api.example.com/",
    bearer_auth=SyncOAuthProvider(
        token_url="https://auth.example.com/oauth/token",
        client_id="my-client-id",
        client_secret="my-client-secret",
    ),
) as client:
    client.get("items/7")

Default headers

For anything that isn't a Bearer token — an API key header, a custom user-agent, whatever your API needs on every request — pass default_headers when creating the client. Unlike bearer_auth, these are fixed for the client's whole lifetime, resolved once, not per-request:

async with HTTPClient(
    base_url="https://api.example.com/",
    default_headers={"x-api-key": "my-api-key"},
) as client:
    await client.get("items/7")  # sent with every request through this client

Combine freely with bearer_token/bearer_auth/basic_auth — they set different headers (Authorization vs. whatever you name here).

Timeouts

timeout (seconds, default 30.0) applies to the whole client, covering every request made through it:

async with HTTPClient(base_url="https://api.example.com/", timeout=5.0) as client:
    await client.get("items/7")  # raises HTTPTimeoutError if this takes longer than 5s

Pass timeout=None to disable it and fall back to pyreqwest's own default. See Error handling for HTTPTimeoutError.

Every verb also takes its own timeout, overriding the client's for that one call only:

async with HTTPClient(base_url="https://api.example.com/", timeout=5.0) as client:
    await client.get("items/7")  # uses the client default, 5s
    await client.get("exports/large-file.csv", timeout=60.0)  # this call gets 60s instead

There's no way to make a single call wait forever when the client itself has a finite timeout — pyreqwest's own per-request .timeout() only ever accepts a duration, never a sentinel meaning "no timeout." To remove the limit entirely, build the client with timeout=None instead.