Spaces:
Running
Running
| from fastapi import APIRouter, Request, Response, status | |
| from fastapi.responses import RedirectResponse, HTMLResponse | |
| from starlette.middleware.sessions import SessionMiddleware | |
| from google_auth_oauthlib.flow import Flow | |
| from googleapiclient.discovery import build | |
| import os | |
| from app.config import GOOGLE_CLIENT_ID, GOOGLE_CLIENT_SECRET, GOOGLE_REDIRECT_URI, YOUTUBE_SCOPES | |
| from app.models import UserDatabase | |
| router = APIRouter() | |
| def create_flow(): | |
| return Flow.from_client_config( | |
| { | |
| "web": { | |
| "client_id": GOOGLE_CLIENT_ID, | |
| "project_id": "youtube-fastapi-sample", | |
| "auth_uri": "https://accounts.google.com/o/oauth2/auth", | |
| "token_uri": "https://oauth2.googleapis.com/token", | |
| "client_secret": GOOGLE_CLIENT_SECRET, | |
| } | |
| }, | |
| scopes=YOUTUBE_SCOPES, | |
| ) | |
| async def login(request: Request): | |
| flow = create_flow() | |
| flow.redirect_uri = GOOGLE_REDIRECT_URI # must set the redirect_uri separately | |
| authorization_url, state = flow.authorization_url( | |
| access_type="offline", | |
| include_granted_scopes="true", | |
| prompt="select_account" | |
| ) | |
| request.session["state"] = state | |
| return RedirectResponse(authorization_url) | |
| async def auth_callback(request: Request): | |
| """Handle OAuth callback from Google with ?code= and ?state=.""" | |
| state = request.session.get("state") | |
| if not state: | |
| return HTMLResponse("<h1>Session state not found. Please /login again.</h1>", status_code=400) | |
| flow = create_flow() | |
| flow.redirect_uri = GOOGLE_REDIRECT_URI # Set the redirect_uri here instead | |
| # Remove the redirect_uri parameter from fetch_token | |
| flow.fetch_token( | |
| authorization_response=str(request.url) | |
| # Don't pass redirect_uri here, as it's already set on the flow object | |
| ) | |
| # Rest of your code remains the same | |
| credentials = flow.credentials | |
| if not credentials or not credentials.valid: | |
| return HTMLResponse("<h1>Invalid credentials. Please /login again.</h1>", status_code=400) | |
| request.session["credentials"] = { | |
| "token": credentials.token, | |
| "refresh_token": credentials.refresh_token, | |
| "token_uri": credentials.token_uri, | |
| "client_id": credentials.client_id, | |
| "client_secret": credentials.client_secret, | |
| "scopes": credentials.scopes | |
| } | |
| youtube = build("youtube", "v3", credentials=credentials) | |
| channel_response = youtube.channels().list(part="snippet", mine=True).execute() | |
| if channel_response.get("items") and len(channel_response["items"]) > 0: | |
| channel_username = channel_response["items"][0]["snippet"]["title"] | |
| else: | |
| channel_username = "unknown_user" | |
| user = UserDatabase.create_user(channel_username, request.session["credentials"]) | |
| response = RedirectResponse(url="/videos", status_code=status.HTTP_302_FOUND) | |
| # Set cookie settings based on your environment | |
| import os | |
| if os.getenv("HF_SPACE") == "true": | |
| secure_cookie = True | |
| samesite_value = "none" | |
| else: | |
| secure_cookie = False | |
| samesite_value = "lax" | |
| # Redirect to /videos with a cookie token | |
| response = RedirectResponse(url="/videos", status_code=303) | |
| response.set_cookie( | |
| key="token", | |
| value=channel_username, | |
| max_age=1800, | |
| httponly=True, | |
| secure=secure_cookie, | |
| samesite=samesite_value | |
| ) | |
| return response | |