File size: 3,278 Bytes
abf702c | 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 | from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi_sso.sso.google import GoogleSSO
from app.core.config import settings
from app.core.auth import create_access_token
from app.db.session import get_db
from app.services.user_service import user_service
from sqlalchemy.ext.asyncio import AsyncSession
from starlette.responses import RedirectResponse
router = APIRouter()
google_sso = GoogleSSO(
client_id=settings.GOOGLE_CLIENT_ID,
client_secret=settings.GOOGLE_CLIENT_SECRET,
redirect_uri=settings.GOOGLE_REDIRECT_URI,
allow_insecure_http=True # Set to False in production
)
@router.get("/google/login")
async def google_login():
with google_sso:
return await google_sso.get_login_redirect()
@router.get("/google/callback")
async def google_callback(request: Request, db: AsyncSession = Depends(get_db)):
with google_sso:
try:
user_data = await google_sso.verify_and_process(request)
except Exception as e:
raise HTTPException(status_code=400, detail=f"Google Authentication failed: {str(e)}")
if not user_data:
raise HTTPException(status_code=400, detail="Failed to get user data from Google")
# Check if user exists
user = await user_service.get_user_by_email(db, user_data.email)
if not user:
user = await user_service.create_user(
db=db,
email=user_data.email,
full_name=user_data.display_name or user_data.email,
avatar_url=user_data.picture,
google_id=user_data.id
)
# Record Session
from app.models.database import UserSession
session_record = UserSession(
user_id=user.id,
ip_address=request.client.host if request.client else None,
user_agent=request.headers.get("user-agent")
)
db.add(session_record)
await db.commit()
# Create access token
access_token = create_access_token(data={"sub": user.email})
# Redirect back to frontend with token
# In a real app, you might use a more secure way to pass the token
import os
frontend_url = os.getenv("FRONTEND_URL", "http://localhost:5173")
return RedirectResponse(url=f"{frontend_url}/auth/callback?token={access_token}")
# Fixing imports and implementing logout properly
from app.core.auth import get_user_or_demo
from app.models.database import UserSession, User as DBUser
from sqlalchemy import select
import datetime
@router.post("/logout")
async def logout_endpoint(db: AsyncSession = Depends(get_db), user: DBUser = Depends(get_user_or_demo)):
# Find the latest open session for this user
result = await db.execute(
select(UserSession)
.where(UserSession.user_id == user.id)
.where(UserSession.logout_at == None)
.order_by(UserSession.login_at.desc())
.limit(1)
)
last_session = result.scalar_one_or_none()
if last_session:
last_session.logout_at = datetime.datetime.utcnow()
await db.commit()
return {"status": "success", "message": "Logged out successfully"}
return {"status": "success", "message": "No active session found"}
|