Skip to content

google_auth

google_auth

Shared Google OAuth helpers: access token read + one-shot 401 refresh.

All Google connectors (Gmail, Calendar, Contacts, Drive, Tasks) authenticate with the same OAuth flow and store identical token payloads at ~/.openjarvis/connectors/*.json — typically a shared google.json file plus per-product copies. They all need the same refresh-on-401 behavior, so the wrapper lives here instead of being duplicated per connector.

Use call_with_refresh(api_fn, credentials_path, *args, **kwargs) around any token-taking API helper. On a 401 the wrapper exchanges the stored refresh_token for a new access_token, updates the credentials file, and retries the call once. All other status codes propagate.

Classes

GoogleAuthError

Bases: RuntimeError

Raised when Google credentials are missing or refresh-token grant fails.

Functions

current_access_token 

current_access_token(credentials_path: str) -> str

Return the current access token from the credentials file (empty if absent).

Source code in src/openjarvis/connectors/google_auth.py 
34
35
36
37
def current_access_token(credentials_path: str) -> str:
    """Return the current access token from the credentials file (empty if absent)."""
    tokens = load_tokens(credentials_path) or {}
    return tokens.get("access_token", tokens.get("token", ""))

refresh_access_token 

refresh_access_token(credentials_path: str) -> str

Exchange the stored refresh_token for a fresh access_token and persist it.

Returns the new access_token. Raises :class:GoogleAuthError when the credentials file is missing, lacks a refresh_token / client credentials, or when Google rejects the refresh grant (e.g. the refresh_token has been revoked and the user needs to re-authenticate).

Source code in src/openjarvis/connectors/google_auth.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def refresh_access_token(credentials_path: str) -> str:
    """Exchange the stored refresh_token for a fresh access_token and persist it.

    Returns the new access_token. Raises :class:`GoogleAuthError` when the
    credentials file is missing, lacks a refresh_token / client credentials,
    or when Google rejects the refresh grant (e.g. the refresh_token has been
    revoked and the user needs to re-authenticate).
    """
    tokens = load_tokens(credentials_path)
    if not tokens:
        raise GoogleAuthError(
            f"No credentials at {credentials_path}; re-run the connector OAuth flow."
        )
    refresh_token = tokens.get("refresh_token", "")
    client_id = tokens.get("client_id", "")
    client_secret = tokens.get("client_secret", "")
    if not (refresh_token and client_id and client_secret):
        raise GoogleAuthError(
            "Stored Google credentials are missing refresh_token / client_id / "
            "client_secret; re-run the connector OAuth flow to mint a full token."
        )

    resp = httpx.post(
        _GOOGLE_TOKEN_ENDPOINT,
        data={
            "client_id": client_id,
            "client_secret": client_secret,
            "refresh_token": refresh_token,
            "grant_type": "refresh_token",
        },
        timeout=30.0,
    )
    if resp.status_code != 200:
        raise GoogleAuthError(
            f"Google token refresh failed ({resp.status_code}): {resp.text[:200]}"
        )
    payload = resp.json()
    new_token = payload.get("access_token", "")
    if not new_token:
        raise GoogleAuthError(
            "Google token refresh returned 200 but no access_token in payload."
        )

    tokens["access_token"] = new_token
    # Keep the legacy "token" key in sync for older code paths that read it.
    tokens["token"] = new_token
    if "expires_in" in payload:
        tokens["expires_in"] = payload["expires_in"]
    save_tokens(credentials_path, tokens)
    logger.info(
        "Refreshed Google access token (expires_in=%s)", payload.get("expires_in")
    )
    return new_token

call_with_refresh 

call_with_refresh(api_fn: Callable[..., Dict[str, Any]], credentials_path: str, *args: Any, **kwargs: Any) -> Dict[str, Any]

Invoke api_fn(token, *args, **kwargs) with one-shot 401 auto-refresh.

Loads the current access token from disk, calls the helper, and if Google returns 401 (the access token has expired or been revoked) uses the stored refresh_token to mint a new access_token, updates the credentials file, and retries the call exactly once.

Any other HTTPStatusError is re-raised unchanged — auth-related retries end here; transient 5xx / timeout retries belong further up the stack.

Source code in src/openjarvis/connectors/google_auth.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def call_with_refresh(
    api_fn: Callable[..., Dict[str, Any]],
    credentials_path: str,
    *args: Any,
    **kwargs: Any,
) -> Dict[str, Any]:
    """Invoke ``api_fn(token, *args, **kwargs)`` with one-shot 401 auto-refresh.

    Loads the current access token from disk, calls the helper, and if Google
    returns 401 (the access token has expired or been revoked) uses the stored
    refresh_token to mint a new access_token, updates the credentials file,
    and retries the call exactly once.

    Any other ``HTTPStatusError`` is re-raised unchanged — auth-related retries
    end here; transient 5xx / timeout retries belong further up the stack.
    """
    token = current_access_token(credentials_path)
    try:
        return api_fn(token, *args, **kwargs)
    except httpx.HTTPStatusError as exc:
        if exc.response is None or exc.response.status_code != 401:
            raise
        logger.info(
            "Google returned 401 on %s — refreshing access token and retrying.",
            getattr(api_fn, "__name__", "<api_fn>"),
        )
        new_token = refresh_access_token(credentials_path)
        return api_fn(new_token, *args, **kwargs)