Core¶
GoogleAuth ¶
GoogleAuth(token_store: TokenStore | None = None, credentials_file: str | None = None, scopes: list[str] | None = None, user_id: str = 'default')
Google OAuth2 authentication handler.
Manages credential acquisition, refresh, and storage. Supports both OAuth (interactive) and Service Account (server-to-server).
Example
OAuth (interactive)¶
auth = GoogleAuth() auth.authenticate() # Opens browser
Service Account¶
auth = GoogleAuth.from_service_account("service-account.json")
Use credentials¶
service = build("gmail", "v1", credentials=auth.credentials)
Initialize OAuth authenticator.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token_store
|
TokenStore | None
|
Storage backend for tokens (default: SQLite) |
None
|
credentials_file
|
str | None
|
Path to OAuth credentials JSON |
None
|
scopes
|
list[str] | None
|
OAuth scopes to request (default: Scopes.default(), i.e. Gmail, Calendar, Drive and Sheets) |
None
|
user_id
|
str
|
User identifier for multi-user setups |
'default'
|
credentials
property
¶
credentials: Credentials | None
Get current credentials, loading from store if needed.
from_service_account
classmethod
¶
from_service_account(service_account_file: str, scopes: list[str] | None = None, subject: str | None = None) -> GoogleAuth
Create authenticator from service account.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
service_account_file
|
str
|
Path to service account JSON |
required |
scopes
|
list[str] | None
|
OAuth scopes |
None
|
subject
|
str | None
|
Email to impersonate (for domain-wide delegation) |
None
|
Returns:
| Type | Description |
|---|---|
GoogleAuth
|
GoogleAuth instance with service account credentials |
refresh ¶
refresh() -> bool
Refresh expired credentials.
Returns:
| Type | Description |
|---|---|
bool
|
True if refresh succeeded |
Raises:
| Type | Description |
|---|---|
TokenRefreshError
|
If refresh fails |
authenticate ¶
authenticate(force: bool = False) -> Credentials
Get valid credentials, running OAuth flow if needed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
force
|
bool
|
Force new authentication even if credentials exist |
False
|
Returns:
| Type | Description |
|---|---|
Credentials
|
Valid Google credentials |
Raises:
| Type | Description |
|---|---|
CredentialsNotFoundError
|
If the browser flow is needed and the credentials file doesn't exist |
revoke ¶
revoke() -> bool
Revoke and delete stored credentials.
Returns:
| Type | Description |
|---|---|
bool
|
True if credentials were deleted |
export_token ¶
export_token() -> dict | None
Export token data for external storage.
Returns:
| Type | Description |
|---|---|
dict | None
|
Token data dict or None if not authenticated |
get_user_email ¶
get_user_email() -> str | None
Get authenticated user's email.
Returns:
| Type | Description |
|---|---|
str | None
|
Email address or None |
Scopes ¶
Google API OAuth scopes.
Use these constants to request specific permissions.
default
classmethod
¶
default() -> list[str]
Default scopes: Gmail, Calendar, Drive and Sheets.
Drive used to be missing, so a default login got 403s from the Drive
client and from the Sheets calls that go through Drive (open by title,
list, share, export). Tokens created before need gsuite auth login --force.
Tasks and Contacts are left out on purpose: contacts is a sensitive
scope, and asking for both on every login would over-ask. Request them
with Scopes.all() or gsuite auth login --scopes all.
Settings ¶
Bases: BaseSettings
Application settings loaded from environment variables.
Prefix: GSUITE_ Example: GSUITE_API_KEY=secret
exceptions ¶
Google Suite exceptions.
GSuiteError ¶
GSuiteError(message: str, cause: Exception | None = None)
Bases: Exception
Base exception for all Google Suite errors.
AuthenticationError ¶
AuthenticationError(message: str, cause: Exception | None = None)
CredentialsNotFoundError ¶
CredentialsNotFoundError(path: str)
TokenExpiredError ¶
TokenExpiredError()
TokenRefreshError ¶
TokenRefreshError(cause: Exception | None = None)
NotAuthenticatedError ¶
NotAuthenticatedError()
APIError ¶
APIError(message: str, service: str, status_code: int | None = None, cause: Exception | None = None)
RateLimitError ¶
RateLimitError(service: str, retry_after: int | None = None)
NotFoundError ¶
NotFoundError(service: str, resource_type: str, resource_id: str)
PermissionDeniedError ¶
PermissionDeniedError(service: str, operation: str)
ValidationError ¶
ValidationError(field: str, message: str)
ConfigurationError ¶
ConfigurationError(message: str, cause: Exception | None = None)
AsyncGoogleClient ¶
AsyncGoogleClient(auth: Any, *, timeout: float | None = None, transport: Any = None)
Authorized async HTTP for Google APIs, with retries and mapped errors.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
auth
|
Any
|
A |
required |
timeout
|
float | None
|
Seconds per request (default: GSUITE_REQUEST_TIMEOUT). |
None
|
transport
|
Any
|
An httpx transport (tests pass |
None
|
request
async
¶
request(method: str, url: str, *, service: str, resource_type: str = 'resource', resource_id: str = 'unknown', params: dict[str, Any] | None = None, json: Any = None, content: bytes | None = None, headers: dict[str, str] | None = None) -> Any
Send a request; returns the httpx.Response of a successful call.
Raises:
| Type | Description |
|---|---|
GSuiteError
|
A subclass of it for HTTP errors. |
TransportError
|
A network error, once retries are exhausted. |
json
async
¶
json(method: str, url: str, **kwargs: Any) -> Any
request() and decode the JSON body ({} when empty).
paginate
async
¶
paginate(url: str, items_key: str, *, service: str, params: dict[str, Any] | None = None, max_items: int | None = None, page_size_param: str = 'maxResults', max_page_size: int = 100) -> AsyncIterator[dict[str, Any]]
Async paginate(): yield items of a GET list endpoint, following nextPageToken.