oauth_manager
Per-user OAuth token management with encrypted storage.
Handles the full OAuth2 authorization-code flow for multiple providers
(GitHub, Google, Discord, Microsoft), stores tokens encrypted at rest
in Redis, and transparently refreshes expired access tokens.
- Redis key pattern:
stargazer:oauth_tokens:{user_id}:{provider} (Fernet-encrypted JSON)
stargazer:oauth_link:{link_code} (one-time link code -> user_id)
-
class oauth_manager.OAuthProvider(name, authorize_url, token_url, client_id='', client_secret='', scopes=<factory>, tokens_expire=True, revoke_url='', extra_auth_params=<factory>)[source]
Bases: object
Configuration for a single OAuth2 provider.
- Parameters:
-
-
name: str
-
authorize_url: str
-
token_url: str
-
client_id: str = ''
-
client_secret: str = ''
-
scopes: list[str]
-
tokens_expire: bool = True
-
revoke_url: str = ''
-
class oauth_manager.TokenData(access_token, refresh_token='', expires_at=0.0, scopes=<factory>, token_type='Bearer', provider='')[source]
Bases: object
Decrypted OAuth token bundle.
- Parameters:
-
-
access_token: str
-
refresh_token: str = ''
-
expires_at: float = 0.0
-
scopes: list[str]
-
token_type: str = 'Bearer'
-
provider: str = ''
-
property is_expired: bool
-
to_dict()[source]
- Return type:
dict[str, Any]
-
classmethod from_dict(d)[source]
- Return type:
TokenData
- Parameters:
d (dict[str, Any])
-
exception oauth_manager.OAuthNotConnected(provider, connect_url)[source]
Bases: Exception
Raised when a tool needs an OAuth token the user hasn’t provided.
- Parameters:
provider (str)
connect_url (str)
- Return type:
None
-
class oauth_manager.OAuthManager(encryption_key='', base_url='', providers_config=None)[source]
Bases: object
Core OAuth2 token lifecycle manager.
- Parameters:
-
-
providers: dict[str, OAuthProvider]
-
is_provider_configured(provider)[source]
- Return type:
bool
- Parameters:
provider (str)
-
list_configured_providers()[source]
- Return type:
list[str]
-
async create_link_code(redis, user_id)[source]
- Return type:
str
- Parameters:
-
-
async resolve_link_code(redis, code)[source]
- Return type:
str | None
- Parameters:
-
-
get_authorize_url(provider, state, scopes=None)[source]
- Return type:
str
- Parameters:
-
-
async exchange_code(provider, code)[source]
- Return type:
TokenData
- Parameters:
-
-
async get_token(user_id, provider, redis)[source]
Return a valid access token, refreshing if needed. Returns None if not connected.
- Return type:
str | None
- Parameters:
-
-
async store_token(redis, user_id, token)[source]
- Return type:
None
- Parameters:
-
-
async delete_token(redis, user_id, provider)[source]
- Return type:
None
- Parameters:
-
-
async list_user_connections(user_id, redis)[source]
Return a list of providers the user has connected.
- Return type:
list[dict[str, Any]]
- Parameters:
-
-
async has_token(user_id, provider, redis)[source]
- Return type:
bool
- Parameters:
-
-
async generate_connect_url(user_id, provider, redis, scopes=None)[source]
- Return type:
str
- Parameters:
-
-
oauth_manager.get_oauth_manager()[source]
Return the global OAuthManager singleton (must be initialized first).
- Return type:
OAuthManager
-
oauth_manager.init_oauth_manager(encryption_key='', base_url='', providers_config=None)[source]
- Return type:
OAuthManager
- Parameters:
-
-
async oauth_manager.require_oauth_token(ctx, provider)[source]
Get a valid token or raise OAuthNotConnected with a connect link.
- Return type:
str
- Parameters:
-