Authentication

See the authentication guide for how these fit together.

The seam

base

The Auth seam and the non-interactive credential providers.

The client never inspects credentials directly: it only ever asks an Auth instance to stamp an outgoing request. The OAuth grants live in datamermaid.auth.oauth, which builds on the same seam.

Classes:

Name	Description
Auth	Base class for credential providers.
AnonymousAuth	Send requests without credentials (public endpoints only).
APIKeyAuth	Authenticate with a MERMAID API key.
HTTPXAuthAdapter	Bridge an Auth into the httpx authentication protocol.

Functions:

Name	Description
resolve_auth	Pick the credential provider to use.

Attributes:

Name	Type	Description
API_KEY_ENV_VAR

API_KEY_ENV_VAR module-attribute

API_KEY_ENV_VAR = 'MERMAID_API_KEY'

Auth

Bases: ABC

Base class for credential providers.

Subclasses implement apply. Implementations that hold expiring credentials additionally override refresh and should_refresh, which together give them a single retry after an authentication failure.

Methods:

Name	Description
apply	Add credentials to request in place.
refresh	Obtain fresh credentials. No-op for static credentials.
should_refresh	Whether response warrants refreshing the credentials and one retry.

apply abstractmethod

apply(request: Request) -> None

Add credentials to request in place.

refresh

refresh() -> None

Obtain fresh credentials. No-op for static credentials.

should_refresh

should_refresh(response: Response) -> bool

Whether response warrants refreshing the credentials and one retry.

The retry goes through refresh. The response body has not been read at this point, so implementations must decide based on the status code and headers only.

AnonymousAuth

Bases: Auth

Send requests without credentials (public endpoints only).

Methods:

Name	Description
apply
refresh	Obtain fresh credentials. No-op for static credentials.
should_refresh	Whether response warrants refreshing the credentials and one retry.

apply

apply(request: Request) -> None

refresh

refresh() -> None

Obtain fresh credentials. No-op for static credentials.

should_refresh

should_refresh(response: Response) -> bool

Whether response warrants refreshing the credentials and one retry.

The retry goes through refresh. The response body has not been read at this point, so implementations must decide based on the status code and headers only.

APIKeyAuth

APIKeyAuth(api_key: str)

Bases: Auth

Authenticate with a MERMAID API key.

The key has the form mmd_<key_id>.<secret> and is sent as Authorization: Bearer <key>.

Methods:

Name	Description
apply
refresh	Obtain fresh credentials. No-op for static credentials.
should_refresh	Whether response warrants refreshing the credentials and one retry.

Attributes:

Name	Type	Description
api_key
key_id	str | None	The public identifier portion of the key, if it is well formed.

api_key instance-attribute

api_key = api_key.strip()

key_id property

key_id: str | None

The public identifier portion of the key, if it is well formed.

apply

apply(request: Request) -> None

refresh

refresh() -> None

Obtain fresh credentials. No-op for static credentials.

should_refresh

should_refresh(response: Response) -> bool

Whether response warrants refreshing the credentials and one retry.

The retry goes through refresh. The response body has not been read at this point, so implementations must decide based on the status code and headers only.

HTTPXAuthAdapter

HTTPXAuthAdapter(auth: Auth)

Bases: Auth

Bridge an Auth into the httpx authentication protocol.

Methods:

Name	Description
auth_flow

Attributes:

Name	Type	Description
auth

auth instance-attribute

auth = auth

auth_flow

auth_flow(
    request: Request,
) -> Generator[Request, Response, None]

resolve_auth

resolve_auth(
    auth: Auth | None = None,
    api_key: str | None = None,
    *,
    env: bool = True,
    cache: bool = True,
) -> Auth

Pick the credential provider to use.

Precedence: explicit auth, then api_key, then the MERMAID_API_KEY environment variable, then tokens left behind by datamermaid.login, then anonymous access.

OAuth

oauth

OAuth 2.0 credentials for MermaidClient.

OAuth owns the boring half of logging in - picking a grant, caching the result, noticing expiry and refreshing - and delegates the interactive half to the flows in datamermaid.auth.flows.

Classes:

Name	Description
OAuth	Log in to MERMAID through Auth0 and keep the token fresh.

Functions:

Name	Description
is_headless	Guess whether this terminal can reach a browser on the same machine.
login	Log in to MERMAID and cache the tokens.
logout	Discard the cached MERMAID tokens. Returns whether any were found.

Attributes:

Name	Type	Description
FlowName
FLOWS	dict[str, type[Flow]]

FlowName module-attribute

FlowName = Literal[
    "auto", "pkce", "implicit", "device", "manual"
]

FLOWS module-attribute

FLOWS: dict[str, type[Flow]] = {
    flow.name: flow
    for flow in (
        PkceFlow,
        ImplicitFlow,
        DeviceFlow,
        ManualPasteFlow,
    )
}

DEFAULT_FLOW_TIMEOUT module-attribute

DEFAULT_FLOW_TIMEOUT = 300.0

OAuth

OAuth(
    *,
    flow: FlowName | Flow = "auto",
    domain: str | None = None,
    client_id: str | None = None,
    audience: str | None = None,
    scope: str | None = None,
    cache: TokenCache | Path | str | bool = True,
    redirect_port: int = 0,
    redirect_host: str = "localhost",
    timeout: float = DEFAULT_FLOW_TIMEOUT,
    interactive: bool = True,
    env: bool = True,
    browser_opener: Callable[[str], bool] | None = None,
    prompt: Callable[[str], str] | None = None,
    printer: Callable[[str], None] | None = None,
    http_client: Client | None = None,
    server_factory: Callable[..., LoopbackCallbackServer]
    | None = None,
    clock: Callable[[], float] | None = None,
    sleep: Callable[[float], None] | None = None,
    now: Callable[[], float] | None = None,
)

Bases: Auth

Log in to MERMAID through Auth0 and keep the token fresh.

The first request triggers a login unless a usable token is already cached. Renewal is single-flight: when several threads sharing one client find the token expired, one of them refreshes it (or logs in) and the others wait for, then use, the result. Pass interactive=False to raise AuthFlowError instead of prompting, which is what an unattended job wants.

Parameters:

Name	Type	Description	Default
flow	FlowName | Flow	Which grant to run, or a Flow instance. "auto" picks pkce on a desktop and device (falling back to manual) on a headless terminal.	'auto'
domain	str | None	Auth0 tenant, e.g. datamermaid.auth0.com.	None
client_id	str | None	Public Auth0 client id.	None
audience	str | None	API identifier the access token must be valid for.	None
scope	str | None	Space-separated OAuth scopes. offline_access is what makes a refresh token available.	None
cache	TokenCache | Path | str | bool	A TokenCache, a path to use instead of the default one, or False to keep tokens in memory only.	True
redirect_port	int	Loopback port for the redirect-based grants. 0 picks an ephemeral one, which needs the tenant to allow port-agnostic callback URLs.	0
redirect_host	str	Host name used to build the redirect URI; the socket always listens on 127.0.0.1.	'localhost'
timeout	float	Seconds to wait for the user to finish the login.	DEFAULT_FLOW_TIMEOUT
interactive	bool	Whether a login may prompt.	True
env	bool	Whether the MERMAID_* environment variables are consulted.	True

Raises:

Type	Description
ValueError	If flow names a grant that does not exist.

Attributes:

Name	Type	Description
config		The resolved Auth0Config.
cache	TokenCache | None	The token cache in use, or None.

Example

from datamermaid import MermaidClient, OAuth

with MermaidClient(auth=OAuth(flow="device")) as client:
    print(client.me().full_name)

Methods:

Name	Description
access_token	A valid access token, refreshing or logging in if need be.
apply
should_refresh	Retry a rejected request once, but only if renewal could help.
refresh	Renew the credentials, falling back to a fresh login.
login	Obtain tokens interactively and cache them.
logout	Forget the cached tokens for this tenant.

config instance-attribute

config = Auth0Config.resolve(
    domain=domain,
    client_id=client_id,
    audience=audience,
    scope=scope,
    env=env,
)

flow instance-attribute

flow = flow

interactive instance-attribute

interactive = interactive

timeout instance-attribute

timeout = timeout

redirect_port instance-attribute

redirect_port = redirect_port

redirect_host instance-attribute

redirect_host = redirect_host

cache instance-attribute

cache: TokenCache | None = cache

tokens property

tokens: TokenSet | None

The token set in memory or in the cache, without logging in.

access_token

access_token() -> str

A valid access token, refreshing or logging in if need be.

apply

apply(request: Request) -> None

should_refresh

should_refresh(response: Response) -> bool

Retry a rejected request once, but only if renewal could help.

refresh

refresh() -> None

Renew the credentials, falling back to a fresh login.

If another thread renewed them after this thread's request was sent, the new token is used as it is, without renewing again.

login

login(*, force: bool = False) -> TokenSet

Obtain tokens interactively and cache them.

A cached token that is still valid (or refreshable) is reused unless force is set. Only one login or refresh runs at a time; a thread that arrives while one is running waits for it.

logout

logout() -> bool

Forget the cached tokens for this tenant.

is_headless

is_headless() -> bool

Guess whether this terminal can reach a browser on the same machine.

True for an SSH session, and for a Unix desktop with no display server.

login

login(*, force: bool = False, **kwargs: Any) -> OAuth

Log in to MERMAID and cache the tokens.

Any keyword OAuth accepts may be passed through, e.g. datamermaid.login(flow="device"). The returned object can be handed to MermaidClient(auth=...), though a plain MermaidClient() will also pick the cached tokens up.

logout

logout(**kwargs: Any) -> bool

Discard the cached MERMAID tokens. Returns whether any were found.

Grants

flows

The interactive OAuth grants used to obtain a MERMAID access token.

Each flow is a small object with a single Flow.authorize method. Everything that would otherwise touch the outside world - the browser, the terminal, the clock, the HTTP client, the loopback server - arrives through a FlowContext, so the tests drive the whole thing without opening a browser or a socket.

Classes:

Name	Description
FlowContext	The outside world a flow is allowed to touch.
Flow	One way of turning a user's consent into a set of tokens.
PkceFlow	Authorization code grant with PKCE, redirected to a loopback server.
ImplicitFlow	Legacy implicit grant: the token arrives in the redirect fragment.
DeviceFlow	Device authorization grant (RFC 8628), for terminals without a browser.
ManualPasteFlow	PKCE without a reachable loopback port: the user pastes the redirect.

Functions:

Name	Description
generate_code_verifier	A fresh PKCE code verifier (unpadded base64url, RFC 7636 section 4.1).
code_challenge	The S256 challenge for verifier.
open_browser	Open url in the system browser, reporting whether that worked.
token_set	Build a token set, reporting a malformed response as a flow error.

Attributes:

Name	Type	Description
DEVICE_GRANT_TYPE

DEVICE_GRANT_TYPE module-attribute

DEVICE_GRANT_TYPE = (
    "urn:ietf:params:oauth:grant-type:device_code"
)

CODE_VERIFIER_BYTES module-attribute

CODE_VERIFIER_BYTES = 32

DEFAULT_DEVICE_INTERVAL module-attribute

DEFAULT_DEVICE_INTERVAL = 5.0

SLOW_DOWN_INCREMENT module-attribute

SLOW_DOWN_INCREMENT = 5.0

FlowContext dataclass

FlowContext(
    http: Client,
    open_browser: Callable[[str], bool] = open_browser,
    prompt: Callable[[str], str] = _default_prompt,
    printer: Callable[[str], None] = print,
    clock: Callable[[], float] = monotonic,
    sleep: Callable[[float], None] = sleep,
    now: Callable[[], float] = time,
    redirect_port: int = 0,
    redirect_host: str = "localhost",
    timeout: float = 300.0,
    server_factory: Callable[
        ..., LoopbackCallbackServer
    ] = LoopbackCallbackServer,
)

The outside world a flow is allowed to touch.

Methods:

Name	Description
open_server

Attributes:

Name	Type	Description
http	Client
open_browser	Callable[[str], bool]
prompt	Callable[[str], str]
printer	Callable[[str], None]
clock	Callable[[], float]
sleep	Callable[[float], None]
now	Callable[[], float]
redirect_port	int
redirect_host	str
timeout	float
server_factory	Callable[..., LoopbackCallbackServer]

http instance-attribute

http: Client

open_browser class-attribute instance-attribute

open_browser: Callable[[str], bool] = open_browser

prompt class-attribute instance-attribute

prompt: Callable[[str], str] = _default_prompt

printer class-attribute instance-attribute

printer: Callable[[str], None] = print

clock class-attribute instance-attribute

clock: Callable[[], float] = time.monotonic

sleep class-attribute instance-attribute

sleep: Callable[[float], None] = time.sleep

now class-attribute instance-attribute

now: Callable[[], float] = time.time

redirect_port class-attribute instance-attribute

redirect_port: int = 0

redirect_host class-attribute instance-attribute

redirect_host: str = 'localhost'

timeout class-attribute instance-attribute

timeout: float = 300.0

server_factory class-attribute instance-attribute

server_factory: Callable[..., LoopbackCallbackServer] = (
    LoopbackCallbackServer
)

open_server

open_server(
    *, mode: str = "query"
) -> LoopbackCallbackServer

Flow

Bases: ABC

One way of turning a user's consent into a set of tokens.

See TokenSet.

Methods:

Name	Description
authorize	Run the grant to completion and return the issued tokens.

Attributes:

Name	Type	Description
name	str

name instance-attribute

name: str

authorize abstractmethod

authorize(
    config: Auth0Config, ctx: FlowContext
) -> TokenSet

Run the grant to completion and return the issued tokens.

PkceFlow

Bases: Flow

Authorization code grant with PKCE, redirected to a loopback server.

Methods:

Name	Description
authorize

Attributes:

Name	Type	Description
name

name class-attribute instance-attribute

name = 'pkce'

authorize

authorize(
    config: Auth0Config, ctx: FlowContext
) -> TokenSet

ImplicitFlow

Bases: Flow

Legacy implicit grant: the token arrives in the redirect fragment.

Kept for parity with the older MERMAID clients. It issues no refresh token, so the user has to log in again when the access token expires.

Methods:

Name	Description
authorize

Attributes:

Name	Type	Description
name

name class-attribute instance-attribute

name = 'implicit'

authorize

authorize(
    config: Auth0Config, ctx: FlowContext
) -> TokenSet

DeviceFlow

DeviceFlow(authorization: dict[str, Any] | None = None)

Bases: Flow

Device authorization grant (RFC 8628), for terminals without a browser.

The user is given a short code and a URL to open on any other device; the terminal polls the token endpoint until they are done.

Methods:

Name	Description
request_device_code	Ask Auth0 for a device and user code.
authorize

Attributes:

Name	Type	Description
name

name class-attribute instance-attribute

name = 'device'

request_device_code

request_device_code(
    config: Auth0Config, ctx: FlowContext
) -> dict[str, Any]

Ask Auth0 for a device and user code.

authorize

authorize(
    config: Auth0Config, ctx: FlowContext
) -> TokenSet

ManualPasteFlow

Bases: Flow

PKCE without a reachable loopback port: the user pastes the redirect.

The browser still lands on http://localhost:<port>/, which fails to connect when the browser is on a different machine - but the address bar holds the authorization code, and pasting that URL back is enough.

Methods:

Name	Description
authorize

Attributes:

Name	Type	Description
name
default_port

name class-attribute instance-attribute

name = 'manual'

default_port class-attribute instance-attribute

default_port = 1410

authorize

authorize(
    config: Auth0Config, ctx: FlowContext
) -> TokenSet

generate_code_verifier

generate_code_verifier(
    *, nbytes: int = CODE_VERIFIER_BYTES
) -> str

A fresh PKCE code verifier (unpadded base64url, RFC 7636 section 4.1).

code_challenge

code_challenge(verifier: str) -> str

The S256 challenge for verifier.

open_browser

open_browser(url: str) -> bool

Open url in the system browser, reporting whether that worked.

Only the failures a browser launch can legitimately produce are caught: a broader except would also swallow the test suite's guard against opening a real browser.

post_form

post_form(
    ctx: FlowContext,
    url: str,
    data: dict[str, str],
    *,
    expect: str = "token",
) -> dict[str, Any]

POST a form to an OAuth endpoint and return the decoded JSON body.

Errors are raised as AuthFlowError carrying Auth0's own error/error_description; the caller sees the response only when it succeeded.

token_set

token_set(
    payload: dict[str, Any], ctx: FlowContext
) -> TokenSet

Build a token set, reporting a malformed response as a flow error.

See TokenSet.

Tenant configuration

config

Auth0 tenant configuration for the OAuth flows.

The defaults point at MERMAID's production Auth0 tenant and are the same values the R client (mermaidr) uses. Every one of them can be overridden per call or through the environment, so the SDK can also be pointed at the development tenant.

Classes:

Name	Description
Auth0Config	Where to send OAuth requests and on whose behalf.

Attributes:

Name	Type	Description
DEFAULT_AUTH0_DOMAIN
DEFAULT_CLIENT_ID
DEFAULT_AUDIENCE
DEFAULT_SCOPE
AUTH0_DOMAIN_ENV_VAR
CLIENT_ID_ENV_VAR
AUDIENCE_ENV_VAR

DEFAULT_AUTH0_DOMAIN module-attribute

DEFAULT_AUTH0_DOMAIN = 'datamermaid.auth0.com'

DEFAULT_CLIENT_ID module-attribute

DEFAULT_CLIENT_ID = '6q1XvYG0n75ZaLbFko0gUV4xGud4uPyG'

DEFAULT_AUDIENCE module-attribute

DEFAULT_AUDIENCE = 'https://api.datamermaid.org'

DEFAULT_SCOPE module-attribute

DEFAULT_SCOPE = 'openid profile email offline_access'

AUTH0_DOMAIN_ENV_VAR module-attribute

AUTH0_DOMAIN_ENV_VAR = 'MERMAID_AUTH0_DOMAIN'

CLIENT_ID_ENV_VAR module-attribute

CLIENT_ID_ENV_VAR = 'MERMAID_CLIENT_ID'

AUDIENCE_ENV_VAR module-attribute

AUDIENCE_ENV_VAR = 'MERMAID_AUDIENCE'

Auth0Config dataclass

Auth0Config(
    domain: str = DEFAULT_AUTH0_DOMAIN,
    client_id: str = DEFAULT_CLIENT_ID,
    audience: str = DEFAULT_AUDIENCE,
    scope: str = DEFAULT_SCOPE,
)

Where to send OAuth requests and on whose behalf.

Methods:

Name	Description
resolve	Build a config from explicit arguments, then the environment, then defaults.
scopes	The scope string, optionally with offline_access removed.

Attributes:

Name	Type	Description
domain	str
client_id	str
audience	str
scope	str
issuer	str
authorize_url	str
token_url	str
device_code_url	str
cache_key	str	Identity of the credentials this config produces, for the token cache.

domain class-attribute instance-attribute

domain: str = DEFAULT_AUTH0_DOMAIN

client_id class-attribute instance-attribute

client_id: str = DEFAULT_CLIENT_ID

audience class-attribute instance-attribute

audience: str = DEFAULT_AUDIENCE

scope class-attribute instance-attribute

scope: str = DEFAULT_SCOPE

issuer property

issuer: str

authorize_url property

authorize_url: str

token_url property

token_url: str

device_code_url property

device_code_url: str

cache_key property

cache_key: str

Identity of the credentials this config produces, for the token cache.

resolve classmethod

resolve(
    *,
    domain: str | None = None,
    client_id: str | None = None,
    audience: str | None = None,
    scope: str | None = None,
    env: bool = True,
) -> Auth0Config

Build a config from explicit arguments, then the environment, then defaults.

scopes

scopes(*, offline_access: bool = True) -> str

The scope string, optionally with offline_access removed.

The implicit grant never issues refresh tokens, and Auth0 rejects the scope outright for it.

Token cache

token_cache

On-disk storage for OAuth tokens.

Tokens live in $XDG_CONFIG_HOME/datamermaid/tokens.json (defaulting to ~/.config), readable only by the owner. One file holds one entry per tenant/client/audience combination, so the production and development Auth0 tenants can be logged into side by side.

Classes:

Name	Description
TokenSet	The credentials returned by a successful OAuth flow.
TokenCache	A JSON file of token sets, one entry per Auth0 cache key.

Functions:

Name	Description
default_cache_path	$XDG_CONFIG_HOME/datamermaid/tokens.json, falling back to ~/.config.

Attributes:

Name	Type	Description
CACHE_FILE_MODE
DEFAULT_LEEWAY

CACHE_VERSION module-attribute

CACHE_VERSION = 1

CACHE_FILE_MODE module-attribute

CACHE_FILE_MODE = 384

CACHE_DIR_MODE module-attribute

CACHE_DIR_MODE = 448

DEFAULT_LEEWAY module-attribute

DEFAULT_LEEWAY = 60.0

TokenSet dataclass

TokenSet(
    access_token: str,
    refresh_token: str | None = None,
    id_token: str | None = None,
    token_type: str = "Bearer",
    scope: str | None = None,
    expires_at: float | None = None,
)

The credentials returned by a successful OAuth flow.

Methods:

Name	Description
from_response	Build a token set from an Auth0 token endpoint response.
from_dict
to_dict
is_expired	Whether the access token is past (or about to pass) its expiry.
merged_with	other on top of this set, keeping a refresh token it omits.

Attributes:

Name	Type	Description
access_token	str
refresh_token	str | None
id_token	str | None
token_type	str
scope	str | None
expires_at	float | None

access_token instance-attribute

access_token: str

refresh_token class-attribute instance-attribute

refresh_token: str | None = None

id_token class-attribute instance-attribute

id_token: str | None = None

token_type class-attribute instance-attribute

token_type: str = 'Bearer'

scope class-attribute instance-attribute

scope: str | None = None

expires_at class-attribute instance-attribute

expires_at: float | None = None

from_response classmethod

from_response(
    payload: dict[str, Any], *, now: float | None = None
) -> TokenSet

Build a token set from an Auth0 token endpoint response.

from_dict classmethod

from_dict(data: dict[str, Any]) -> TokenSet

to_dict

to_dict() -> dict[str, Any]

is_expired

is_expired(
    *,
    leeway: float = DEFAULT_LEEWAY,
    now: float | None = None,
) -> bool

Whether the access token is past (or about to pass) its expiry.

A token with no known expiry is assumed valid: the API is the authority, and a 401 triggers a refresh anyway.

merged_with

merged_with(other: TokenSet) -> TokenSet

other on top of this set, keeping a refresh token it omits.

Auth0 does not return the refresh token again when rotation is off, so a naive replace would throw away the only way to renew silently.

TokenCache

TokenCache(path: Path | str | None = None)

A JSON file of token sets, one entry per Auth0 cache key.

The key is Auth0Config.cache_key.

save and clear each read the file, change it and write it back while holding a lock on a .lock file beside it, so two clients or processes updating different entries never lose each other's changes.

Methods:

Name	Description
load	The cached tokens for key, or None.
save	Store tokens under key with 0600 permissions.
clear	Forget key (or every entry when it is None).

Attributes:

Name	Type	Description
path
lock_path	Path	The lock file that serialises updates to path.

path instance-attribute

path = (
    Path(path) if path is not None else default_cache_path()
)

lock_path property

lock_path: Path

The lock file that serialises updates to path.

load

load(key: str) -> TokenSet | None

The cached tokens for key, or None.

save

save(key: str, tokens: TokenSet) -> None

Store tokens under key with 0600 permissions.

clear

clear(key: str | None = None) -> bool

Forget key (or every entry when it is None).

Returns whether anything was removed.

default_cache_path

default_cache_path() -> Path

$XDG_CONFIG_HOME/datamermaid/tokens.json, falling back to ~/.config.

Redirect server

callback_server

A single-shot loopback HTTP server for OAuth redirects.

The browser is sent to http://localhost:<port>/ after the user approves the login. For the authorization code grant the parameters arrive in the query string and can be read directly; for the implicit grant they arrive in the URL fragment, which never reaches the server, so the page served there bounces them back as a query string.

Classes:

Name	Description
LoopbackCallbackServer	Serve exactly one OAuth redirect on 127.0.0.1.

Functions:

Name	Description
parse_fragment	Parse the parameters of an OAuth redirect fragment.

Attributes:

Name	Type	Description
CallbackMode
FRAGMENT_PATH
HANDLER_TIMEOUT

CallbackMode module-attribute

CallbackMode = Literal['query', 'fragment']

FRAGMENT_PATH module-attribute

FRAGMENT_PATH = '/fragment'

BIND_HOST module-attribute

BIND_HOST = '127.0.0.1'

DEFAULT_REDIRECT_HOST module-attribute

DEFAULT_REDIRECT_HOST = 'localhost'

DEFAULT_TIMEOUT module-attribute

DEFAULT_TIMEOUT = 300.0

HANDLER_TIMEOUT module-attribute

HANDLER_TIMEOUT = 30.0

LoopbackCallbackServer

LoopbackCallbackServer(
    *,
    mode: CallbackMode = "query",
    port: int = 0,
    redirect_host: str = DEFAULT_REDIRECT_HOST,
    path: str = "/",
    timeout: float = DEFAULT_TIMEOUT,
    clock: Callable[[], float] = monotonic,
)

Serve exactly one OAuth redirect on 127.0.0.1.

The socket is bound as soon as the object is created, so the ephemeral port (and therefore redirect_uri) is known before the authorization URL is built.

Methods:

Name	Description
wait	Block until the browser hits the callback, or time out.
close

Attributes:

Name	Type	Description
timeout
redirect_host
path
port	int
redirect_uri	str

timeout instance-attribute

timeout = timeout

redirect_host instance-attribute

redirect_host = redirect_host

path instance-attribute

path = path if path.startswith('/') else f'/{path}'

port property

port: int

redirect_uri property

redirect_uri: str

wait

wait() -> dict[str, str]

Block until the browser hits the callback, or time out.

close

close() -> None

parse_fragment

parse_fragment(value: str) -> dict[str, str]

Parse the parameters of an OAuth redirect fragment.

Accepts a whole redirect URL, a bare #access_token=... fragment, or the fragment's contents.

parse_redirect

parse_redirect(value: str) -> dict[str, str]

Parse the parameters of a pasted redirect URL, from either half.

The authorization code grant answers in the query string and the implicit grant in the fragment, so both are considered; the fragment wins when a URL somehow carries both.

Token decoding

jwt

Just enough JWT handling to know when an access token has expired.

The SDK never trusts a token it decodes here: the API verifies signatures. All that is needed locally is the exp claim, so the payload is base64 decoded without any signature check and without a crypto dependency.

Functions:

Name	Description
decode_payload	Return the claims of token, or {} if it is not a readable JWT.
token_expires_at	The exp claim as a POSIX timestamp, or None if there is none.

decode_payload

decode_payload(token: str) -> dict[str, Any]

Return the claims of token, or {} if it is not a readable JWT.

The signature is neither present nor checked; opaque tokens simply yield an empty mapping.

token_expires_at

token_expires_at(token: str) -> float | None

The exp claim as a POSIX timestamp, or None if there is none.