feat(api): Google provider module — centralized external API gateway (#43)

Add Google API authentication and service factory as the first
non-LLM provider in API's evolution to centralized external service
gateway (DPLAN-0036).

New handlers: google/auth.py (OAuth2 lifecycle), google/service_factory.py
(single + thread-safe service objects), google/retry.py (SSL retry).
New module: google_client.py with get_drive_service(), get_google_service(),
validate_google(), reauth_google() public API.

New commands: drone @api validate google, drone @api reauth google.
Credentials at ~/.secrets/aipass/ (cross-platform standard).

Co-authored-by: @api <api@aipass>
This commit is contained in:
AIPass
2026-03-14 23:07:00 -07:00
committed by GitHub
co-authored by @api
parent 4c43610816
commit 9ab96013e7
8 changed files with 869 additions and 49 deletions
+9 -40
View File
@@ -2,45 +2,14 @@
## Identity
API is the **LLM access layer** for AIPass. All branches route through here for model calls, key management, and usage tracking. Gateway to OpenRouter (and future providers).
API is the **centralized external API gateway** for AIPass. Provides authenticated service clients for external APIs. Consumers import ready-to-use clients — API owns the plumbing, consumers own the business logic.
## Commands
## Key Breadcrumbs
```
drone @api get-key [provider] # Retrieve API key (fallback: ~/.secrets/aipass/.env → config → env)
drone @api validate [provider] # Validate key format + connectivity
drone @api test # Test OpenRouter connection
drone @api models # List available models
drone @api track # Track usage metrics
drone @api stats # Usage statistics
```
## Architecture
3-tier: Entry point (`apps/api.py`) → Modules (3) → Handlers (9 files, 6 domains)
**Modules:** `api_key.py` (key mgmt), `openrouter_client.py` (LLM client), `usage_tracker.py` (metrics)
**Handlers:** `auth/` (keys, env), `config/` (provider), `openrouter/` (client, models, caller, provision), `usage/` (tracking, aggregation, cleanup), `json/` (auto-creating JSON ops)
## Cross-Branch API
```python
from aipass.api.apps.modules.openrouter_client import get_response
response = get_response(prompt="...", model="anthropic/claude-3.5-sonnet", caller="flow")
```
Used by: flow, prax, skills. Callers must specify model — no default (intentional).
## Key Files
- `apps/api.py` — Entry point with auto-discovery
- `apps/handlers/auth/keys.py` — Key fallback chain (config → env → .env)
- `apps/handlers/auth/env.py` — .env search paths (priority: `~/.secrets/aipass/`)
- `api_json/` — Auto-created JSON storage (config, data, logs)
## Memory & Tracking
- `.trinity/` — Identity, session history, observations
- `dev.local.md` — Working notes, todos, friction
- `logs/` — Prax log output
- **Credentials live at** `~/.secrets/aipass/` — `google_creds.json`, `google_client_secret.json`, `.env`
- **Design rule:** If it's not auth, credentials, or service factory — it doesn't belong here. See DPLAN-0036 for the full rationale and old Telegram anti-pattern.
- **Provider pattern:** One module per provider (`openrouter_client.py`, `google_client.py`), one handler directory per provider (`openrouter/`, `google/`). Module orchestrates, handlers implement.
- **No default models/configs** — consumers provide their own. API provides the connection.
- **Thread-safe mode:** `get_drive_service(thread_safe=True)` loads fresh creds from disk per call for concurrent workers.
- **Google libs are optional deps** — guarded by `GOOGLE_AUTH_AVAILABLE` flag, commands fail explicitly with install instructions.
- **After building:** Run `drone @seedgo audit aipass @api` before reporting complete.
+51 -9
View File
@@ -1,23 +1,29 @@
# API
**Purpose:** LLM API access layer with provider abstraction, key management, model routing, and usage tracking.
**Purpose:** Centralized external API gateway — authenticated service clients for all external APIs (OpenRouter, Google, future providers).
**Module:** `aipass.api`
**Last Updated:** 2026-03-08
**Last Updated:** 2026-03-14
---
## Overview
### What I Do
- Manage API keys for LLM providers (currently OpenRouter)
- Route requests to LLM models with provider-level abstraction
- Track API usage metrics and provide statistics
- Validate credentials and test provider connectivity
- Discover available models from configured providers
- Provide authenticated service clients for external APIs (Google Drive, OpenRouter, etc.)
- Manage OAuth2 flows, credential storage, and token refresh
- Offer thread-safe service factories for concurrent consumers
- Handle API key management and validation across providers
- Provide SSL retry and connection resilience utilities
### What I Don't Do
- Host business logic — consumers own what they do with the service
- Set default models or configs — consumers provide their own
- Manage application workflows, polling loops, or orchestration
### How I Work
- **Entry Point:** `apps/api.py` -- auto-discovers and routes to modules
- **Pattern:** Standard AIPass 3-tier architecture (entry point / modules / handlers)
- **Design principle:** If it's not auth, credentials, or service factory — it doesn't belong here
---
@@ -26,6 +32,8 @@
```bash
drone @api get-key # Retrieve API key for provider
drone @api validate # Validate API credentials and connection
drone @api validate google # Validate Google OAuth2 credentials
drone @api reauth google # Re-authenticate Google OAuth2
drone @api test # Test OpenRouter connection status
drone @api models # List available models from provider
drone @api track # Track API usage metrics
@@ -38,6 +46,28 @@ Running `drone @api` with no arguments displays module introspection (discovered
---
## Cross-Branch API
```python
# LLM access (OpenRouter)
from aipass.api.apps.modules.openrouter_client import get_response
response = get_response(prompt="...", model="anthropic/claude-3.5-sonnet", caller="flow")
# Google Drive (or any Google API)
from aipass.api.apps.modules.google_client import get_drive_service
service = get_drive_service() # Single-threaded
service = get_drive_service(thread_safe=True) # For concurrent workers
# Any Google service
from aipass.api.apps.modules.google_client import get_google_service
service = get_google_service("calendar", "v3")
# Retry utility for raw API calls
from aipass.api.apps.modules.google_client import api_call_with_retry
```
---
## Architecture
```
@@ -48,6 +78,7 @@ api/
│ ├── modules/
│ │ ├── api_key.py # Key retrieval and validation logic
│ │ ├── openrouter_client.py # OpenRouter API client
│ │ ├── google_client.py # Google API services (Drive, Calendar, etc.)
│ │ └── usage_tracker.py # Usage metrics tracking
│ └── handlers/
│ ├── auth/
@@ -55,6 +86,10 @@ api/
│ │ └── keys.py # API key storage and retrieval
│ ├── config/
│ │ └── provider.py # Provider configuration management
│ ├── google/
│ │ ├── auth.py # OAuth2 lifecycle, credential I/O
│ │ ├── service_factory.py # Service object factory (single + thread-safe)
│ │ └── retry.py # SSL retry with exponential backoff
│ ├── json/
│ │ └── json_handler.py # JSON operation logging
│ ├── openrouter/
@@ -78,9 +113,16 @@ api/
- `aipass.cli` -- Rich console output formatting
### Provides To
- All modules -- LLM API access for any branch that needs model inference
- All branches -- authenticated external API clients
- `@backup` -- Google Drive service (migration from self-contained auth)
- `@skills` -- future external API integrations (Telegram, Google services)
- System-wide API key management and credential validation
### Credentials
- `~/.secrets/aipass/.env` -- API keys (OpenRouter, etc.)
- `~/.secrets/aipass/google_creds.json` -- Google OAuth2 tokens
- `~/.secrets/aipass/google_client_secret.json` -- Google OAuth app config
---
*Last Updated: 2026-03-08*
*Last Updated: 2026-03-14*
+2
View File
@@ -153,6 +153,8 @@ def print_help():
table.add_row("get-key", "Retrieve API key for provider")
table.add_row("validate", "Validate API credentials and connection")
table.add_row("validate google", "Validate Google OAuth2 credentials")
table.add_row("reauth google", "Re-authenticate Google OAuth2")
table.add_row("test", "Test OpenRouter connection status")
table.add_row("models", "List available models from provider")
table.add_row("track", "Track API usage metrics")
@@ -0,0 +1,12 @@
"""
Google API Domain
Handlers for Google service authentication, credential management,
and service object factories. Provides authenticated clients for
Google APIs (Drive, Calendar, etc.) to consuming branches.
"""
__version__ = "1.0.0"
from . import auth as auth
from . import service_factory as service_factory
from . import retry as retry
+243
View File
@@ -0,0 +1,243 @@
# =================== AIPass ====================
# Name: auth.py
# Description: Google OAuth2 authentication and credential management
# Version: 1.0.0
# Created: 2026-03-14
# Modified: 2026-03-14
# =============================================
"""
Google OAuth2 Authentication Handler
Manages the OAuth2 lifecycle for Google API access:
- Load/save credentials from ~/.secrets/aipass/
- Token refresh for expired credentials
- Full OAuth2 consent flow for new authentication
- Re-authentication when tokens are revoked
This is pure auth plumbing — no business logic.
Consumers get authenticated credentials, they decide what to do with them.
"""
from pathlib import Path
from typing import Optional
# =============================================
# CONSTANTS
# =============================================
# Default scopes per service — consumers can override
DEFAULT_SCOPES = {
"drive": ["https://www.googleapis.com/auth/drive.file"],
"calendar": ["https://www.googleapis.com/auth/calendar.readonly"],
}
# Credential storage — AIPass standard location
SECRETS_DIR = Path.home() / ".secrets" / "aipass"
CREDS_PATH = SECRETS_DIR / "google_creds.json"
CLIENT_SECRET_PATH = SECRETS_DIR / "google_client_secret.json"
# =============================================
# GOOGLE API AVAILABILITY
# =============================================
try:
from google.oauth2.credentials import Credentials
from google.auth.transport.requests import Request
from google_auth_oauthlib.flow import InstalledAppFlow
GOOGLE_AUTH_AVAILABLE = True
except ImportError:
GOOGLE_AUTH_AVAILABLE = False
Credentials = None # type: ignore[assignment, misc]
Request = None # type: ignore[assignment, misc]
InstalledAppFlow = None # type: ignore[assignment, misc]
# =============================================
# CREDENTIAL OPERATIONS
# =============================================
def is_available() -> bool:
"""Check if Google auth libraries are installed."""
return GOOGLE_AUTH_AVAILABLE
def load_credentials(scopes: Optional[list] = None) -> Optional["Credentials"]:
"""Load saved OAuth2 credentials from disk.
Args:
scopes: OAuth2 scopes to validate against.
Defaults to Drive scopes if not provided.
Returns:
Credentials object if found and loadable, None otherwise.
"""
if not GOOGLE_AUTH_AVAILABLE:
return None
if not CREDS_PATH.exists():
return None
effective_scopes = scopes or DEFAULT_SCOPES["drive"]
try:
creds = Credentials.from_authorized_user_file(str(CREDS_PATH), effective_scopes)
return creds
except Exception:
return None
def refresh_credentials(creds: "Credentials") -> bool:
"""Attempt to refresh expired credentials.
Args:
creds: Expired Credentials object with a refresh token.
Returns:
True if refresh succeeded, False otherwise.
"""
if not GOOGLE_AUTH_AVAILABLE:
return False
if not creds or not creds.expired or not creds.refresh_token:
return False
try:
creds.refresh(Request())
_save_credentials(creds)
return True
except Exception:
return False
def run_oauth_flow(
scopes: Optional[list] = None,
port: int = 0,
open_browser: bool = True,
) -> Optional["Credentials"]:
"""Run the full OAuth2 consent flow.
Requires google_client_secret.json at ~/.secrets/aipass/.
Opens a local server for the OAuth callback.
Args:
scopes: OAuth2 scopes to request.
port: Local server port (0 = auto-assign).
open_browser: Whether to auto-open the consent page.
Returns:
Credentials object if flow succeeded, None otherwise.
"""
if not GOOGLE_AUTH_AVAILABLE:
return None
if not CLIENT_SECRET_PATH.exists():
return None
effective_scopes = scopes or DEFAULT_SCOPES["drive"]
try:
flow = InstalledAppFlow.from_client_secrets_file(
str(CLIENT_SECRET_PATH), effective_scopes
)
creds = flow.run_local_server(port=port, open_browser=open_browser)
_save_credentials(creds)
return creds
except Exception:
return None
def authenticate(scopes: Optional[list] = None) -> Optional["Credentials"]:
"""Full authentication lifecycle: load → refresh → OAuth flow.
Tries in order:
1. Load existing valid credentials
2. Refresh expired credentials
3. Run full OAuth2 consent flow
Args:
scopes: OAuth2 scopes. Defaults to Drive scopes.
Returns:
Valid Credentials object, or None if all methods fail.
"""
if not GOOGLE_AUTH_AVAILABLE:
return None
# Step 1: Load existing
creds = load_credentials(scopes)
if creds and creds.valid:
return creds
# Step 2: Refresh expired
if creds and creds.expired and creds.refresh_token:
if refresh_credentials(creds):
return creds
# Step 3: Full OAuth flow
return run_oauth_flow(scopes=scopes)
def reauth(
scopes: Optional[list] = None,
port: int = 8085,
open_browser: bool = False,
) -> Optional["Credentials"]:
"""Force re-authentication via OAuth flow (console mode).
Used when existing credentials are revoked or corrupted.
Defaults to console-friendly settings (no browser, fixed port).
Args:
scopes: OAuth2 scopes.
port: Local server port for callback.
open_browser: Whether to auto-open browser.
Returns:
Fresh Credentials object, or None on failure.
"""
# Try refresh first — maybe token just expired
creds = load_credentials(scopes)
if creds and creds.expired and creds.refresh_token:
if refresh_credentials(creds):
return creds
# Force new flow
return run_oauth_flow(scopes=scopes, port=port, open_browser=open_browser)
def validate_credentials(scopes: Optional[list] = None) -> bool:
"""Check if valid Google credentials exist.
Args:
scopes: OAuth2 scopes to validate against.
Returns:
True if valid (or refreshable) credentials exist.
"""
creds = load_credentials(scopes)
if not creds:
return False
if creds.valid:
return True
if creds.expired and creds.refresh_token:
return refresh_credentials(creds)
return False
# =============================================
# INTERNAL HELPERS
# =============================================
def _save_credentials(creds: "Credentials") -> None:
"""Save credentials to the standard secrets path."""
SECRETS_DIR.mkdir(parents=True, exist_ok=True)
with open(CREDS_PATH, "w", encoding="utf-8") as f:
f.write(creds.to_json())
@@ -0,0 +1,88 @@
# =================== AIPass ====================
# Name: retry.py
# Description: Google API retry logic with SSL error handling
# Version: 1.0.0
# Created: 2026-03-14
# Modified: 2026-03-14
# =============================================
"""
Google API Retry Handler
Provides exponential backoff retry for Google API calls,
with specific handling for transient SSL/connection errors.
Extracted from backup's drive_sync_client.py — generic enough
for any Google API consumer, not just Drive.
Usage:
from aipass.api.apps.handlers.google.retry import api_call_with_retry
result = api_call_with_retry(
service.files().list(q="..."),
max_retries=3,
rebuild_service_fn=my_rebuild_fn,
)
"""
import ssl
import time
from typing import Any, Callable, Optional
def is_ssl_error(exc: Exception) -> bool:
"""Check if an exception is a transient SSL/connection error.
Args:
exc: The caught exception.
Returns:
True if the error is a transient SSL/connection issue.
"""
if isinstance(exc, (ssl.SSLError, BrokenPipeError, ConnectionResetError)):
return True
ssl_keywords = (
"DECRYPTION_FAILED_OR_BAD_RECORD_MAC",
"WRONG_VERSION_NUMBER",
"EOF occurred",
"ssl.SSLError",
"BrokenPipeError",
"ConnectionReset",
)
msg = str(exc)
return any(kw in msg for kw in ssl_keywords)
def api_call_with_retry(
request: Any,
max_retries: int = 3,
rebuild_service_fn: Optional[Callable] = None,
) -> Any:
"""Execute a Google API request with exponential backoff on SSL errors.
Args:
request: A Google API request object (has .execute() method).
max_retries: Maximum number of retry attempts.
rebuild_service_fn: Optional callback to rebuild the service
on SSL failure (e.g. to get a fresh connection).
Called with no arguments, return value is ignored.
Returns:
The API response.
Raises:
The original exception if retries are exhausted or
the error is not an SSL/connection issue.
"""
for attempt in range(max_retries + 1):
try:
return request.execute()
except Exception as e:
if attempt < max_retries and is_ssl_error(e):
wait = 2 ** attempt
time.sleep(wait)
if rebuild_service_fn:
rebuild_service_fn()
continue
raise
@@ -0,0 +1,119 @@
# =================== AIPass ====================
# Name: service_factory.py
# Description: Google API service object factory
# Version: 1.0.0
# Created: 2026-03-14
# Modified: 2026-03-14
# =============================================
"""
Google API Service Factory
Builds authenticated Google API service objects (Drive, Calendar, etc.).
Supports both single-threaded and thread-safe modes.
Thread-safe mode loads fresh credentials from disk per call,
avoiding token refresh races in concurrent operations.
This pattern was extracted from backup's drive_sync_client.py.
Usage:
from aipass.api.apps.handlers.google.service_factory import (
build_service, build_thread_safe_service
)
# Single-threaded
service = build_service("drive", "v3")
# Thread-safe (for concurrent workers)
service = build_thread_safe_service("drive", "v3")
"""
from typing import Optional
from aipass.api.apps.handlers.google import auth as auth
# =============================================
# GOOGLE API AVAILABILITY
# =============================================
try:
from googleapiclient.discovery import build
GOOGLE_BUILD_AVAILABLE = True
except ImportError:
GOOGLE_BUILD_AVAILABLE = False
build = None # type: ignore[assignment]
# =============================================
# SERVICE FACTORIES
# =============================================
def build_service(
service_name: str = "drive",
version: str = "v3",
scopes: Optional[list] = None,
) -> Optional[object]:
"""Build an authenticated Google API service object.
Uses the full auth lifecycle (load → refresh → OAuth flow).
Args:
service_name: Google API service (e.g. "drive", "calendar", "sheets").
version: API version (e.g. "v3", "v3").
scopes: OAuth2 scopes. Defaults to service-specific defaults from auth module.
Returns:
Authenticated service object, or None if auth/build fails.
"""
if not GOOGLE_BUILD_AVAILABLE or not auth.is_available():
return None
creds = auth.authenticate(scopes=scopes)
if not creds:
return None
try:
return build(service_name, version, credentials=creds)
except Exception:
return None
def build_thread_safe_service(
service_name: str = "drive",
version: str = "v3",
scopes: Optional[list] = None,
) -> Optional[object]:
"""Build an isolated service instance for use in a worker thread.
Loads fresh credentials from disk to avoid sharing credential state
(token refresh races) and creates a fully isolated HTTP/SSL connection.
Args:
service_name: Google API service name.
version: API version.
scopes: OAuth2 scopes.
Returns:
Isolated authenticated service object, or None on failure.
"""
if not GOOGLE_BUILD_AVAILABLE or not auth.is_available():
return None
# Load fresh credentials from disk — no shared state
creds = auth.load_credentials(scopes=scopes)
if not creds:
return None
# Refresh if expired
if creds.expired and creds.refresh_token:
if not auth.refresh_credentials(creds):
return None
if not creds.valid:
return None
try:
return build(service_name, version, credentials=creds)
except Exception:
return None
@@ -0,0 +1,345 @@
# =================== AIPass ====================
# Name: google_client.py
# Description: Google API Client Module — public API for Google services
# Version: 1.0.0
# Created: 2026-03-14
# Modified: 2026-03-14
# =============================================
"""
Google API Client Module
Public API for Google service access across AIPass.
Consumers import from here — never directly from handlers.
Provides:
- get_drive_service() → Authenticated Google Drive v3 client
- get_google_service() → Any Google API service (Calendar, Sheets, etc.)
- authenticate_google() → Run OAuth2 flow and return credentials
- validate_google() → Check if valid credentials exist
- reauth_google() → Force re-authentication
Consumer pattern:
from aipass.api.apps.modules.google_client import get_drive_service
service = get_drive_service()
service.files().list(...).execute()
Thread-safe pattern (for concurrent workers):
service = get_drive_service(thread_safe=True)
"""
import sys
from typing import List, Optional
from aipass.prax.apps.modules.logger import system_logger as logger # noqa: F811
from aipass.cli.apps.modules import console, header, success, error, warning
from aipass.api.apps.handlers.json import json_handler
import aipass.api.apps.handlers.google.auth as google_auth
import aipass.api.apps.handlers.google.service_factory as google_factory
import aipass.api.apps.handlers.google.retry as google_retry
# =============================================
# MODULE INTROSPECTION
# =============================================
def print_introspection() -> None:
"""Show module introspection — connected handlers and capabilities."""
console.print()
header("Google Client Module Introspection")
console.print()
console.print("[cyan]Purpose:[/cyan] Google API authentication and service factory")
console.print()
console.print("[cyan]Connected Handlers:[/cyan]")
console.print(" - api.apps.handlers.google.auth")
console.print(" - api.apps.handlers.google.service_factory")
console.print(" - api.apps.handlers.google.retry")
console.print()
console.print("[cyan]Available Workflows:[/cyan]")
console.print(" - get_drive_service() - Get authenticated Drive client")
console.print(" - get_google_service() - Get any Google API service")
console.print(" - authenticate_google() - Run OAuth2 authentication")
console.print(" - validate_google() - Check credential status")
console.print(" - reauth_google() - Force re-authentication")
console.print()
available = google_auth.is_available()
status = "[green]installed[/green]" if available else "[red]missing[/red]"
console.print(f"[cyan]Google Libraries:[/cyan] {status}")
has_creds = google_auth.CREDS_PATH.exists()
cred_status = "[green]found[/green]" if has_creds else "[yellow]not configured[/yellow]"
console.print(f"[cyan]Credentials:[/cyan] {cred_status}")
has_secret = google_auth.CLIENT_SECRET_PATH.exists()
secret_status = "[green]found[/green]" if has_secret else "[yellow]not configured[/yellow]"
console.print(f"[cyan]Client Secret:[/cyan] {secret_status}")
console.print()
def print_help() -> None:
"""Print module help."""
import argparse
parser = argparse.ArgumentParser(
prog="drone @api",
description="Google Client - Google API authentication and service access",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
COMMANDS (via drone @api):
validate google - Check Google OAuth2 credentials
reauth google - Re-run OAuth2 flow for Google
CROSS-BRANCH API:
from aipass.api.apps.modules.google_client import get_drive_service
service = get_drive_service()
CREDENTIAL SETUP:
1. Get OAuth client secret from Google Cloud Console
2. Save as: ~/.secrets/aipass/google_client_secret.json
3. Run: drone @api reauth google
4. Complete OAuth consent in browser
5. Credentials saved to: ~/.secrets/aipass/google_creds.json
"""
)
console.print(parser.format_help())
# =============================================
# COMMAND HANDLING (drone @api validate google, etc.)
# =============================================
def handle_command(command: str, args: List[str]) -> bool:
"""Handle Google client commands routed via drone.
Args:
command: Command name (e.g. "validate", "reauth")
args: Command arguments — first arg should be "google"
Returns:
True if command was handled, False to pass through.
"""
# Only handle commands with "google" as the provider argument
if not args or args[0] != "google":
return False
if command == "validate":
_cmd_validate()
return True
elif command == "reauth":
_cmd_reauth()
return True
return False
# =============================================
# CLI COMMAND IMPLEMENTATIONS
# =============================================
def _cmd_validate() -> None:
"""Validate Google OAuth2 credentials."""
header("Validate Google Credentials")
console.print()
if not google_auth.is_available():
error(
"Google auth libraries not installed",
suggestion="pip install google-auth google-auth-oauthlib google-api-python-client",
)
return
if not google_auth.CLIENT_SECRET_PATH.exists():
error(
"Client secret not found",
suggestion=f"Save OAuth client secret to: {google_auth.CLIENT_SECRET_PATH}",
)
return
if google_auth.validate_credentials():
success("Google credentials are valid")
json_handler.log_operation("google_validate", {"status": "valid"})
else:
warning("No valid Google credentials found")
console.print()
console.print("[dim]Run 'drone @api reauth google' to authenticate[/dim]")
json_handler.log_operation("google_validate", {"status": "invalid"})
def _cmd_reauth() -> None:
"""Force Google re-authentication via OAuth flow."""
header("Google Re-Authentication")
console.print()
if not google_auth.is_available():
error(
"Google auth libraries not installed",
suggestion="pip install google-auth google-auth-oauthlib google-api-python-client",
)
return
if not google_auth.CLIENT_SECRET_PATH.exists():
error(
"Client secret not found",
suggestion=f"Save OAuth client secret to: {google_auth.CLIENT_SECRET_PATH}",
)
return
warning("Starting OAuth2 flow...")
console.print("[dim]A browser window may open for Google consent.[/dim]")
console.print()
creds = google_auth.reauth()
if creds:
success("Google re-authentication successful")
console.print(f"[dim]Credentials saved to: {google_auth.CREDS_PATH}[/dim]")
json_handler.log_operation("google_reauth", {"status": "success"})
else:
error("Google re-authentication failed")
json_handler.log_operation("google_reauth", {"status": "failed"})
# =============================================
# PUBLIC API — Cross-branch imports
# =============================================
def get_drive_service(thread_safe: bool = False) -> object:
"""Get an authenticated Google Drive v3 service object.
Args:
thread_safe: If True, builds an isolated service instance
with fresh credentials from disk (for concurrent workers).
Returns:
Authenticated Drive v3 service object.
Raises:
RuntimeError: If authentication fails or libraries unavailable.
"""
return get_google_service("drive", "v3", thread_safe=thread_safe)
def get_google_service(
service_name: str = "drive",
version: str = "v3",
scopes: Optional[list] = None,
thread_safe: bool = False,
) -> object:
"""Get an authenticated Google API service object.
Supports any Google API: Drive, Calendar, Sheets, Gmail, etc.
Args:
service_name: Google API service (e.g. "drive", "calendar").
version: API version (e.g. "v3").
scopes: OAuth2 scopes. Uses service-specific defaults if not provided.
thread_safe: If True, builds an isolated instance for concurrent use.
Returns:
Authenticated service object.
Raises:
RuntimeError: If authentication fails or libraries unavailable.
"""
if not google_auth.is_available():
raise RuntimeError(
"Google auth libraries not installed. "
"Install: pip install google-auth google-auth-oauthlib google-api-python-client"
)
if thread_safe:
service = google_factory.build_thread_safe_service(
service_name, version, scopes
)
else:
service = google_factory.build_service(service_name, version, scopes)
if not service:
raise RuntimeError(
f"Failed to authenticate with Google {service_name} API. "
"Run 'drone @api reauth google' to set up credentials."
)
return service
def authenticate_google(scopes: Optional[list] = None) -> bool:
"""Run Google OAuth2 authentication.
Args:
scopes: OAuth2 scopes to request.
Returns:
True if authentication succeeded.
"""
creds = google_auth.authenticate(scopes=scopes)
return creds is not None
def validate_google(scopes: Optional[list] = None) -> bool:
"""Check if valid Google credentials exist.
Args:
scopes: OAuth2 scopes to validate against.
Returns:
True if valid credentials exist.
"""
return google_auth.validate_credentials(scopes=scopes)
def reauth_google(scopes: Optional[list] = None) -> bool:
"""Force Google re-authentication.
Args:
scopes: OAuth2 scopes to request.
Returns:
True if re-authentication succeeded.
"""
creds = google_auth.reauth(scopes=scopes)
return creds is not None
# Re-export retry utility for consumers that make raw API calls
api_call_with_retry = google_retry.api_call_with_retry
is_ssl_error = google_retry.is_ssl_error
# =============================================
# STANDALONE EXECUTION
# =============================================
if __name__ == "__main__":
args = sys.argv[1:]
if len(args) == 0:
print_introspection()
sys.exit(0)
if args[0] in ["--help", "-h", "help"]:
print_help()
sys.exit(0)
command = args[0]
remaining_args = args[1:] if len(args) > 1 else []
if handle_command(command, remaining_args):
sys.exit(0)
else:
console.print()
console.print(f"[red]Unknown command: {command}[/red]")
console.print()
console.print(
"Run [dim]python3 google_client.py --help[/dim] for available commands"
)
console.print()
sys.exit(1)