src/google_oauth.py

from urllib.parse import urlencode

import requests

_AUTHORIZE_URL = "https://accounts.google.com/o/oauth2/v2/auth"
_TOKEN_URL = "https://oauth2.googleapis.com/token"
_DRIVE_FILE_SCOPE = "https://www.googleapis.com/auth/drive.file"


def build_authorize_url(client_id: str, redirect_uri: str, state: str) -> str:
    """Builds the Google consent-screen URL for the one-time Drive connection.

    access_type=offline + prompt=consent force Google to include a
    refresh_token in the token response, even for a client that has been
    authorized before.
    """
    params = {
        "client_id": client_id,
        "redirect_uri": redirect_uri,
        "response_type": "code",
        "scope": _DRIVE_FILE_SCOPE,
        "access_type": "offline",
        "prompt": "consent",
        "state": state,
    }
    return f"{_AUTHORIZE_URL}?{urlencode(params)}"


def exchange_code(client_id: str, client_secret: str, redirect_uri: str, code: str) -> dict:
    """Exchanges an authorization code for tokens, including a refresh_token."""
    response = requests.post(
        _TOKEN_URL,
        data={
            "client_id": client_id,
            "client_secret": client_secret,
            "redirect_uri": redirect_uri,
            "code": code,
            "grant_type": "authorization_code",
        },
        timeout=10,
    )
    response.raise_for_status()
    return response.json()


def refresh_access_token(client_id: str, client_secret: str, refresh_token: str) -> dict:
    """Mints a fresh access_token from a stored refresh_token, no browser needed.

    Raises requests.HTTPError (e.g. invalid_grant) if the refresh_token has
    been revoked -- callers should delete the stored token and prompt the
    user to reconnect Drive.
    """
    response = requests.post(
        _TOKEN_URL,
        data={
            "client_id": client_id,
            "client_secret": client_secret,
            "refresh_token": refresh_token,
            "grant_type": "refresh_token",
        },
        timeout=10,
    )
    response.raise_for_status()
    return response.json()