""" Google OAuth2 flow implementation. Isolated OAuth2 client that can run without FastAPI. """ from typing import Optional from google.auth.transport.requests import Request from google.oauth2.credentials import Credentials from google_auth_oauthlib.flow import Flow from .config import settings class GoogleOAuth: """ Google OAuth2 client. Handles authorization flow, token exchange, and token refresh. """ def __init__( self, client_id: Optional[str] = None, client_secret: Optional[str] = None, redirect_uri: Optional[str] = None, scopes: Optional[list[str]] = None, ): """ Initialize OAuth client. Falls back to settings if parameters not provided. """ self.client_id = client_id or settings.google_client_id self.client_secret = client_secret or settings.google_client_secret self.redirect_uri = redirect_uri or settings.google_redirect_uri self.scopes = scopes or settings.google_scopes.split() def _create_flow(self) -> Flow: """Create OAuth flow object.""" client_config = { "web": { "client_id": self.client_id, "client_secret": self.client_secret, "auth_uri": "https://accounts.google.com/o/oauth2/auth", "token_uri": "https://oauth2.googleapis.com/token", } } flow = Flow.from_client_config( client_config, scopes=self.scopes, redirect_uri=self.redirect_uri, ) return flow def get_authorization_url(self, state: Optional[str] = None) -> str: """ Generate OAuth2 authorization URL. Args: state: Optional state parameter for CSRF protection Returns: URL to redirect user for Google authorization """ flow = self._create_flow() auth_url, _ = flow.authorization_url( access_type="offline", # Request refresh token include_granted_scopes="true", state=state, ) return auth_url def exchange_code_for_tokens(self, code: str) -> dict: """ Exchange authorization code for tokens. Args: code: Authorization code from callback Returns: Token dict containing: - access_token - refresh_token - expires_in - scope - token_type """ flow = self._create_flow() flow.fetch_token(code=code) credentials = flow.credentials return { "access_token": credentials.token, "refresh_token": credentials.refresh_token, "expires_in": 3600, # Google tokens typically 1 hour "scope": " ".join(credentials.scopes or []), "token_type": "Bearer", } def refresh_access_token(self, refresh_token: str) -> dict: """ Refresh an expired access token. Args: refresh_token: The refresh token Returns: New token dict with fresh access_token """ credentials = Credentials( token=None, refresh_token=refresh_token, token_uri="https://oauth2.googleapis.com/token", client_id=self.client_id, client_secret=self.client_secret, ) request = Request() credentials.refresh(request) return { "access_token": credentials.token, "refresh_token": refresh_token, # Keep original refresh token "expires_in": 3600, "scope": " ".join(credentials.scopes or []), "token_type": "Bearer", } def get_credentials(self, access_token: str, refresh_token: Optional[str] = None) -> Credentials: """ Create Google Credentials object from tokens. Args: access_token: OAuth2 access token refresh_token: Optional refresh token Returns: Google Credentials object for API calls """ return Credentials( token=access_token, refresh_token=refresh_token, token_uri="https://oauth2.googleapis.com/token", client_id=self.client_id, client_secret=self.client_secret, scopes=self.scopes, )