OAuth Flows

class Flow:
    """OAuth 2.0 Authorization Flow for Google APIs."""
    
    def __init__(
        self,
        oauth2session: requests_oauthlib.OAuth2Session,
        client_type: str,
        client_config: Mapping[str, Any],
        redirect_uri: Optional[str] = None,
        code_verifier: Optional[str] = None,
        autogenerate_code_verifier: bool = True
    ):
        """
        Initialize OAuth 2.0 flow.

        Args:
            oauth2session: OAuth 2.0 session from requests-oauthlib
            client_type: Either "web" or "installed"
            client_config: Client configuration in Google client secrets format
            redirect_uri: OAuth 2.0 redirect URI (optional at creation time)
            code_verifier: PKCE code verifier string (43-128 chars)
            autogenerate_code_verifier: Auto-generate PKCE code verifier
        """
    
    @classmethod
    def from_client_config(
        cls, 
        client_config: Mapping[str, Any], 
        scopes: Sequence[str], 
        **kwargs
    ) -> "Flow":
        """
        Create Flow from client configuration dictionary.

        Args:
            client_config: Client configuration in Google client secrets format
            scopes: List of OAuth 2.0 scopes to request
            **kwargs: Additional parameters for OAuth2Session

        Returns:
            Configured Flow instance

        Raises:
            ValueError: If client configuration format is invalid
        """
    
    @classmethod
    def from_client_secrets_file(
        cls, 
        client_secrets_file: str, 
        scopes: Sequence[str], 
        **kwargs
    ) -> "Flow":
        """
        Create Flow from client secrets JSON file.

        Args:
            client_secrets_file: Path to client secrets .json file
            scopes: List of OAuth 2.0 scopes to request
            **kwargs: Additional parameters for OAuth2Session

        Returns:
            Configured Flow instance
        """
    
    def authorization_url(self, **kwargs) -> Tuple[str, str]:
        """
        Generate authorization URL for user consent.

        First step in OAuth 2.0 flow. User's browser should visit returned URL.

        Args:
            **kwargs: Additional parameters for authorization URL

        Returns:
            Tuple of (authorization_url, state) where state is used for verification
        """
    
    def fetch_token(self, **kwargs) -> Mapping[str, str]:
        """
        Complete authorization flow and obtain access token.

        Final step after user grants consent. Exchanges authorization code for tokens.

        Args:
            **kwargs: Must include either 'code' or 'authorization_response'

        Returns:
            Token dictionary with access_token, refresh_token, etc.
        """
    
    @property
    def credentials(self) -> google.oauth2.credentials.Credentials:
        """
        OAuth 2.0 credentials from the session.

        Returns:
            Google Auth credentials object

        Raises:
            ValueError: If no access token available (call fetch_token first)
        """
    
    @property
    def redirect_uri(self) -> str:
        """OAuth 2.0 redirect URI."""
    
    @redirect_uri.setter
    def redirect_uri(self, value: str):
        """Set OAuth 2.0 redirect URI."""
    
    def authorized_session(self) -> google.auth.transport.requests.AuthorizedSession:
        """
        Returns requests session authorized with credentials.

        Returns:
            Authorized session for making API calls
        """

class InstalledAppFlow(Flow):
    """Authorization flow for installed applications."""
    
    def run_local_server(
        self,
        host: str = "localhost",
        bind_addr: Optional[str] = None,
        port: int = 8080,
        authorization_prompt_message: Optional[str] = None,
        success_message: str = "The authentication flow has completed.",
        open_browser: bool = True,
        redirect_uri_trailing_slash: bool = True,
        timeout_seconds: Optional[int] = None,
        token_audience: Optional[str] = None,
        browser: Optional[str] = None,
        **kwargs
    ) -> google.oauth2.credentials.Credentials:
        """
        Run complete OAuth flow using local web server.

        Starts local server, opens browser for user authorization, handles
        redirect, exchanges code for tokens, and returns credentials.

        Args:
            host: Hostname for local redirect server
            bind_addr: IP address to bind server (defaults to host)
            port: Port for local redirect server
            authorization_prompt_message: Message shown to user (None to suppress)
            success_message: Message displayed in browser after completion
            open_browser: Whether to automatically open browser
            redirect_uri_trailing_slash: Include trailing slash in redirect URI
            timeout_seconds: Server timeout (None for no timeout)
            token_audience: Token audience for access token
            browser: Specific browser to use (None for default)
            **kwargs: Additional parameters for authorization_url

        Returns:
            OAuth 2.0 credentials for authenticated user

        Raises:
            Various exceptions for network, OAuth, or configuration errors
        """

from google_auth_oauthlib.flow import Flow

# Load client configuration
client_config = {
    "web": {
        "client_id": "your-client-id.apps.googleusercontent.com",
        "client_secret": "your-client-secret",
        "auth_uri": "https://accounts.google.com/o/oauth2/auth",
        "token_uri": "https://oauth2.googleapis.com/token",
        "redirect_uris": ["http://localhost:8080/callback"]
    }
}

# Create flow
flow = Flow.from_client_config(
    client_config,
    scopes=['https://www.googleapis.com/auth/cloud-platform']
)
flow.redirect_uri = 'http://localhost:8080/callback'

# Step 1: Get authorization URL
auth_url, state = flow.authorization_url(
    access_type='offline',
    include_granted_scopes='true'
)

print(f"Visit this URL to authorize: {auth_url}")

# Step 2: After user authorization, exchange code for token
authorization_code = input("Enter authorization code: ")
flow.fetch_token(code=authorization_code)

# Get credentials
credentials = flow.credentials

from google_auth_oauthlib.flow import InstalledAppFlow

# Create flow from client secrets file
flow = InstalledAppFlow.from_client_secrets_file(
    'client_secrets.json',
    scopes=['https://www.googleapis.com/auth/drive.readonly']
)

# Run complete flow with local server
credentials = flow.run_local_server(
    port=8080,
    open_browser=True
)

# Use credentials
session = flow.authorized_session()
response = session.get('https://www.googleapis.com/drive/v3/files')

# Configure local server behavior
credentials = flow.run_local_server(
    host='127.0.0.1',
    port=9090,
    authorization_prompt_message="Please visit this URL: {url}",
    success_message="Authorization complete! You may close this window.",
    open_browser=False,
    timeout_seconds=300
)

# Flow automatically handles PKCE for enhanced security
flow = Flow.from_client_config(
    client_config,
    scopes=scopes,
    code_verifier='custom-verifier-string'  # Or auto-generate
)

# PKCE parameters automatically included in authorization URL
auth_url, state = flow.authorization_url()

# Flow instance attributes
flow.client_type: str  # "web" or "installed"
flow.client_config: Mapping[str, Any]  # OAuth client configuration
flow.oauth2session: requests_oauthlib.OAuth2Session  # Underlying session
flow.redirect_uri: str  # OAuth redirect URI
flow.code_verifier: Optional[str]  # PKCE code verifier

# Default messages for InstalledAppFlow
InstalledAppFlow._DEFAULT_AUTH_PROMPT_MESSAGE: str
InstalledAppFlow._DEFAULT_AUTH_CODE_MESSAGE: str
InstalledAppFlow._DEFAULT_WEB_SUCCESS_MESSAGE: str