Spaces:
Sleeping
Sleeping
File size: 2,086 Bytes
76b596f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 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 | 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()
|