migrated all pawprint work

artery/veins/google/core/oauth.py

@@ -0,0 +1,147 @@
+"""
+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,
+        )