File size: 5,137 Bytes
db83efb c75e17a db83efb c75e17a db83efb c75e17a 4fe2f95 c75e17a 4fe2f95 c75e17a 4fe2f95 c75e17a 4fe2f95 c75e17a |
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 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 93 94 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 123 124 |
"""Module for defining the main routes of the API."""
import os
import json
import pickle
from fastapi import APIRouter, Request, HTTPException
from fastapi.responses import JSONResponse
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from oauthlib.oauth2.rfc6749.errors import InvalidGrantError
router = APIRouter(tags=["auth"])
CLIENT_ID = os.environ.get("CLIENT_ID")
CLIENT_SECRET = os.environ.get("CLIENT_SECRET")
REDIRECT_URI = os.environ.get("REDIRECT_URI")
SCOPES = ["https://www.googleapis.com/auth/gmail.readonly"]
# Client config for OAuth flow
CLIENT_CONFIG = {
"web": {
"client_id": CLIENT_ID,
"client_secret": CLIENT_SECRET,
"redirect_uris": [REDIRECT_URI],
"auth_uri": "https://accounts.google.com/o/oauth2/auth",
"token_uri": "https://oauth2.googleapis.com/token",
}
}
@router.get("/auth/google/url")
async def get_auth_url():
"""
Handles the generation of a Google OAuth 2.0 authorization URL.
This endpoint initializes an OAuth 2.0 flow using the provided client configuration
and scopes, sets the redirect URI, and generates an authorization URL for the user
to grant access.
Returns:
dict: A dictionary containing the generated authorization URL under the key "url".
"""
flow = InstalledAppFlow.from_client_config(CLIENT_CONFIG, SCOPES)
flow.redirect_uri = REDIRECT_URI
auth_url, _ = flow.authorization_url(access_type="offline", prompt="consent")
return JSONResponse({"url": auth_url})
@router.get("/auth/google/callback")
async def google_callback(code: str, state: str = None, scope: str = None, request: Request = None):
try:
# Validate state (optional, for CSRF protection)
if state and request.state.session.get("oauth_state") != state:
raise HTTPException(status_code=400, detail="Invalid state parameter")
flow = InstalledAppFlow.from_client_config(CLIENT_CONFIG, SCOPES)
flow.redirect_uri = REDIRECT_URI
flow.fetch_token(code=code)
credentials = flow.credentials
request.state.session["credential"] = json.loads(credentials.to_json())
service = build("gmail", "v1", credentials=credentials)
profile = service.users().getProfile(userId="me").execute()
# Ensure cache directory exists
os.makedirs("cache", exist_ok=True)
with open(f"cache/{profile['emailAddress']}.pickle", "wb") as token:
pickle.dump(credentials, token)
return JSONResponse(profile)
except InvalidGrantError:
raise HTTPException(status_code=400, detail="Invalid or expired authorization code")
except Exception as e:
raise HTTPException(status_code=500, detail=f"Authentication failed: {str(e)}")
# @router.get("/auth/google/callback")
# async def google_callback(code: str, request: Request):
# """
# Handles the Google OAuth2 callback by exchanging the authorization code for credentials,
# retrieving the user's Gmail profile, and saving the credentials to a file.
# Args:
# code (str): The authorization code returned by Google's OAuth2 server.
# request (Request): The incoming HTTP request object.
# Returns:
# JSONResponse: A JSON response containing the user's Gmail profile information.
# Side Effects:
# - Saves the user's credentials to a pickle file named after their email address.
# - Stores the credentials in the session state of the request.
# Dependencies:
# - google_auth_oauthlib.flow.InstalledAppFlow: Used to handle the OAuth2 flow.
# - googleapiclient.discovery.build: Used to build the Gmail API service.
# - json: Used to serialize and deserialize credentials.
# - pickle: Used to save credentials to a file.
# """
# flow = InstalledAppFlow.from_client_config(CLIENT_CONFIG, SCOPES)
# flow.redirect_uri = REDIRECT_URI
# flow.fetch_token(code=code)
# credentials = flow.credentials
# request.state.session["credential"] = json.loads(credentials.to_json())
# # cred_dict = (request.state.session.get("credential"))
# # cred = Credentials(
# # token=cred_dict["token"],
# # refresh_token=cred_dict["refresh_token"],
# # token_uri=cred_dict["token_uri"],
# # client_id=cred_dict["client_id"],
# # client_secret=cred_dict["client_secret"],
# # scopes=cred_dict["scopes"],
# # )
# # service = build("gmail", "v1", credentials=Credentials(
# # token=cred_dict["token"],
# # refresh_token=cred_dict["refresh_token"],
# # token_uri=cred_dict["token_uri"],
# # client_id=cred_dict["client_id"],
# # client_secret=cred_dict["client_secret"],
# # scopes=cred_dict["scopes"],
# # ))
# service = build("gmail", "v1", credentials=credentials)
# profile = service.users().getProfile(userId="me").execute()
# with open(f"cache/{profile['emailAddress']}.pickle", "wb") as token:
# pickle.dump(credentials, token)
# return JSONResponse(profile)
|