Spaces:
Running
Running
File size: 3,573 Bytes
1256bd3 b0f2d3d 1256bd3 b0f2d3d 1256bd3 b0f2d3d 1256bd3 904f598 1256bd3 904f598 1256bd3 904f598 1bbee3f 904f598 1bbee3f 904f598 1256bd3 904f598 1256bd3 b0f2d3d 904f598 b0f2d3d |
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 |
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,
)
@router.get("/auth")
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)
@router.get("/auth/callback")
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
|