pywry.auth

OAuth2 authentication system — providers, token storage, session management, and flow orchestration.

---

Package Exports

OAuth2 authentication system for PyWry.

Provides OAuth2 provider abstractions, token storage, session management, and auth flow orchestration for both native window and deploy modes.

Classes

AuthFlowManager

AuthFlowManager(provider: OAuthProvider, token_store: TokenStore | None = None, session_manager: SessionManager | None = None, use_pkce: bool = True, auth_timeout: float = 120.0)

Orchestrates OAuth2 authentication flows.

Supports two modes:

• Native mode: Opens a dedicated auth window pointing at the provider's authorize URL, captures the redirect on an ephemeral localhost HTTP server, and exchanges the code for tokens.

• Deploy mode: Returns the /auth/login URL for the frontend to navigate to. The server-side routes handle the actual flow.

Initialize the auth flow manager.

Attributes

flow_state property

flow_state: AuthFlowState

Current state of the auth flow.

Functions

run_native

run_native(show_window: Any | None = None, close_window: Any | None = None, window_config: dict[str, Any] | None = None) -> AuthFlowResult

Run the OAuth2 flow in native window mode.

This method blocks until authentication completes, times out, or is cancelled.

RAISES	DESCRIPTION
AuthFlowTimeout	If the callback is not received within the timeout.
AuthFlowCancelled	If the user closes the auth window or aborts.
AuthenticationError	If the provider returns an error.

run_deploy

run_deploy(base_url: str = '') -> str

Get the login URL for deploy mode.

In deploy mode, the actual OAuth2 flow is handled server-side by the routes in deploy_routes.py. This method simply returns the URL that the frontend should navigate to.

authenticate

authenticate(app: Any, show_window: Any | None = None, close_window: Any | None = None, window_config: dict[str, Any] | None = None) -> AuthFlowResult

Unified authentication entry point.

Detects the app mode and runs the appropriate flow.

cancel

cancel() -> None

Cancel the current authentication flow.

Sets the cancellation event which unblocks _wait_for_callback_with_cancellation.

OAuthProvider

OAuthProvider(client_id: str, client_secret: str = '', scopes: list[str] | None = None, authorize_url: str = '', token_url: str = '', userinfo_url: str = '', revocation_url: str = '')

Bases: ABC

Abstract base class for OAuth2 providers.

Initialize OAuth provider.

Functions

close async

close() -> None

Close the shared HTTP client. Call from app shutdown lifecycle.

build_authorize_url

build_authorize_url(redirect_uri: str, state: str, pkce: PKCEChallenge | None = None, extra_params: dict[str, str] | None = None) -> str

Build the full authorization URL.

exchange_code abstractmethod async

exchange_code(code: str, redirect_uri: str, pkce_verifier: str | None = None, nonce: str | None = None) -> OAuthTokenSet

Exchange authorization code for tokens.

RAISES	DESCRIPTION
TokenError	If the code exchange fails.

refresh_tokens abstractmethod async

refresh_tokens(refresh_token: str) -> OAuthTokenSet

Refresh an expired access token.

RAISES	DESCRIPTION
TokenRefreshError	If the refresh fails.

get_userinfo async

get_userinfo(access_token: str) -> dict[str, Any]

Fetch user profile information from the provider.

revoke_token async

revoke_token(token: str) -> bool

Revoke a token at the provider (RFC 7009).

Posts to the revocation_url if one is configured. Subclasses with non-standard revocation APIs should override.

GenericOIDCProvider

GenericOIDCProvider(client_id: str, client_secret: str = '', issuer_url: str = '', scopes: list[str] | None = None, authorize_url: str = '', token_url: str = '', userinfo_url: str = '', revocation_url: str = '', *, require_id_token_validation: bool = True)

Bases: OAuthProvider

Generic OpenID Connect provider.

Supports auto-discovery from /.well-known/openid-configuration if an issuer_url is provided.

Initialize OIDC provider.

Functions

close async

close() -> None

Close the shared HTTP client. Call from app shutdown lifecycle.

build_authorize_url

build_authorize_url(redirect_uri: str, state: str, pkce: PKCEChallenge | None = None, extra_params: dict[str, str] | None = None) -> str

Build the full authorization URL.

get_userinfo async

get_userinfo(access_token: str) -> dict[str, Any]

Fetch user profile information from the provider.

validate_id_token async

validate_id_token(id_token: str, nonce: str | None = None) -> dict[str, Any]

Validate an OIDC ID token.

Checks signature (via JWKS), issuer, audience, expiry, and nonce.

RAISES	DESCRIPTION
TokenError	If validation fails for any reason.

exchange_code async

exchange_code(code: str, redirect_uri: str, pkce_verifier: str | None = None, nonce: str | None = None) -> OAuthTokenSet

Exchange authorization code for tokens via OIDC token endpoint.

refresh_tokens async

refresh_tokens(refresh_token: str) -> OAuthTokenSet

Refresh tokens via OIDC token endpoint.

revoke_token async

revoke_token(token: str) -> bool

Revoke a token via the OIDC revocation endpoint.

Triggers endpoint auto-discovery before delegating to the base implementation.

GoogleProvider

GoogleProvider(client_id: str, client_secret: str = '', scopes: list[str] | None = None)

Bases: GenericOIDCProvider

Google OAuth2 provider with preset endpoints.

Initialize Google provider.

Functions

close async

close() -> None

Close the shared HTTP client. Call from app shutdown lifecycle.

exchange_code async

exchange_code(code: str, redirect_uri: str, pkce_verifier: str | None = None, nonce: str | None = None) -> OAuthTokenSet

Exchange authorization code for tokens via OIDC token endpoint.

refresh_tokens async

refresh_tokens(refresh_token: str) -> OAuthTokenSet

Refresh tokens via OIDC token endpoint.

get_userinfo async

get_userinfo(access_token: str) -> dict[str, Any]

Fetch user profile information from the provider.

revoke_token async

revoke_token(token: str) -> bool

Revoke a token via the OIDC revocation endpoint.

Triggers endpoint auto-discovery before delegating to the base implementation.

validate_id_token async

validate_id_token(id_token: str, nonce: str | None = None) -> dict[str, Any]

Validate an OIDC ID token.

Checks signature (via JWKS), issuer, audience, expiry, and nonce.

RAISES	DESCRIPTION
TokenError	If validation fails for any reason.

build_authorize_url

build_authorize_url(redirect_uri: str, state: str, pkce: PKCEChallenge | None = None, extra_params: dict[str, str] | None = None) -> str

Build Google authorization URL with access_type=offline.

GitHubProvider

GitHubProvider(client_id: str, client_secret: str = '', scopes: list[str] | None = None)

Bases: OAuthProvider

GitHub OAuth2 provider.

GitHub uses a non-standard token exchange (no OIDC) and a separate API endpoint for user info.

Initialize GitHub provider.

Functions

close async

close() -> None

Close the shared HTTP client. Call from app shutdown lifecycle.

build_authorize_url

build_authorize_url(redirect_uri: str, state: str, pkce: PKCEChallenge | None = None, extra_params: dict[str, str] | None = None) -> str

Build the full authorization URL.

get_userinfo async

get_userinfo(access_token: str) -> dict[str, Any]

Fetch user profile information from the provider.

exchange_code async

exchange_code(code: str, redirect_uri: str, pkce_verifier: str | None = None, nonce: str | None = None) -> OAuthTokenSet

Exchange authorization code for a GitHub access token.

refresh_tokens async

refresh_tokens(refresh_token: str) -> OAuthTokenSet

Refresh GitHub tokens (GitHub uses token rotation).

revoke_token async

revoke_token(token: str) -> bool

Revoke a GitHub token via the Applications API.

Uses DELETE /applications/{client_id}/token with HTTP Basic authentication (client_id / client_secret).

MicrosoftProvider

MicrosoftProvider(client_id: str, client_secret: str = '', tenant_id: str = 'common', scopes: list[str] | None = None)

Bases: GenericOIDCProvider

Microsoft / Azure AD OAuth2 provider.

Initialize Microsoft provider.

Functions

close async

close() -> None

Close the shared HTTP client. Call from app shutdown lifecycle.

build_authorize_url

build_authorize_url(redirect_uri: str, state: str, pkce: PKCEChallenge | None = None, extra_params: dict[str, str] | None = None) -> str

Build the full authorization URL.

exchange_code async

exchange_code(code: str, redirect_uri: str, pkce_verifier: str | None = None, nonce: str | None = None) -> OAuthTokenSet

Exchange authorization code for tokens via OIDC token endpoint.

refresh_tokens async

refresh_tokens(refresh_token: str) -> OAuthTokenSet

Refresh tokens via OIDC token endpoint.

get_userinfo async

get_userinfo(access_token: str) -> dict[str, Any]

Fetch user profile information from the provider.

revoke_token async

revoke_token(token: str) -> bool

Revoke a token via the OIDC revocation endpoint.

Triggers endpoint auto-discovery before delegating to the base implementation.

validate_id_token async

validate_id_token(id_token: str, nonce: str | None = None) -> dict[str, Any]

Validate an OIDC ID token.

Checks signature (via JWKS), issuer, audience, expiry, and nonce.

RAISES	DESCRIPTION
TokenError	If validation fails for any reason.

SessionManager

SessionManager(provider: OAuthProvider, token_store: TokenStore, session_key: str = 'default', session_store: SessionStore | None = None, refresh_buffer_seconds: int = 60, on_reauth_required: Callable[[], None] | None = None)

Manages OAuth2 token lifecycle with automatic refresh.

Initialize the session manager.

Functions

initialize async

initialize() -> OAuthTokenSet | None

Load existing tokens from store and validate.

save_tokens async

save_tokens(tokens: OAuthTokenSet) -> None

Persist tokens and schedule background refresh.

get_access_token async

get_access_token() -> str

Get a valid access token, refreshing if near expiry.

RAISES	DESCRIPTION
TokenExpiredError	If the token is expired and cannot be refreshed.

refresh async

refresh() -> OAuthTokenSet

Refresh the access token.

Uses the stored refresh token to obtain a new access token. Falls back to on_reauth_required if refresh fails.

RAISES	DESCRIPTION
TokenRefreshError	If refresh fails and no re-auth callback is set.

logout async

logout() -> None

Clear all tokens and cancel scheduled refresh.

Optionally revokes the token at the provider.

TokenStore

Bases: ABC

Abstract base class for OAuth2 token storage.

All methods are async to support both local and network-backed stores.

Functions

save abstractmethod async

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

Save tokens under the given key.

load abstractmethod async

load(key: str) -> OAuthTokenSet | None

Load tokens for the given key.

delete abstractmethod async

delete(key: str) -> None

Delete tokens for the given key.

exists abstractmethod async

exists(key: str) -> bool

Check if tokens exist for the given key.

list_keys abstractmethod async

list_keys() -> list[str]

List all stored token keys.

MemoryTokenStore

MemoryTokenStore()

Bases: TokenStore

In-memory token store for development and single-process use.

Thread-safe via asyncio.Lock, same pattern as MemorySessionStore.

Initialize the memory token store.

Functions

save async

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

Save tokens in memory.

load async

load(key: str) -> OAuthTokenSet | None

Load tokens from memory.

delete async

delete(key: str) -> None

Delete tokens from memory.

exists async

exists(key: str) -> bool

Check if tokens exist in memory.

list_keys async

list_keys() -> list[str]

List all token keys in memory.

RedisTokenStore

RedisTokenStore(redis_url: str = 'redis://localhost:6379/0', prefix: str = 'pywry', pool_size: int = 10)

Bases: TokenStore

Redis-backed token store for multi-worker deployments.

Uses Redis hash keys with optional TTL based on token expires_in.

Initialize the Redis token store.

Functions

save async

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

Save tokens to Redis with optional TTL.

load async

load(key: str) -> OAuthTokenSet | None

Load tokens from Redis.

delete async

delete(key: str) -> None

Delete tokens from Redis.

exists async

exists(key: str) -> bool

Check if tokens exist in Redis.

list_keys async

list_keys() -> list[str]

List all token keys in Redis.

PKCEChallenge dataclass

PKCEChallenge(verifier: str, challenge: str, method: str = 'S256')

PKCE code verifier and challenge pair.

ATTRIBUTE	DESCRIPTION
verifier	The code verifier (high-entropy random string). TYPE: str
challenge	The code challenge (base64url-encoded SHA-256 hash of verifier). TYPE: str
method	The challenge method, always "S256". TYPE: str

Functions

generate classmethod

generate(length: int = 64) -> PKCEChallenge

Generate a new PKCE code verifier and challenge.

Functions

create_provider_from_settings

create_provider_from_settings(settings: Any) -> OAuthProvider

Create an OAuthProvider instance from OAuth2Settings.

RAISES	DESCRIPTION
AuthenticationError	If the provider type is unknown or settings are invalid.

get_token_store

get_token_store(backend: str = 'memory', **kwargs: Any) -> TokenStore

Factory function for token stores.

Returns a singleton instance. Call reset_token_store() to clear the cached instance (e.g. in tests).

---

Submodules

Module	Description
Providers	OAuthProvider ABC and concrete implementations (Google, GitHub, Microsoft, OIDC)
Token Store	Pluggable token storage backends (memory, keyring, Redis)
Flow	AuthFlowManager — orchestrates the full OAuth2 authorization code flow
Session	SessionManager — token lifecycle with automatic background refresh
Deploy Routes	FastAPI router for server-side OAuth2 in deploy mode
Callback Server	Ephemeral localhost server for capturing OAuth2 redirects
PKCE	RFC 7636 PKCE code challenge generation