| from fastapi import APIRouter, Depends, HTTPException, Response, Cookie, Request |
| from fastapi.responses import RedirectResponse |
| from sqlalchemy.ext.asyncio import AsyncSession |
| from pydantic import BaseModel |
| from typing import Optional |
| import os |
|
|
| from app.database import get_db |
| from app.oauth import oauth_handler |
| from app.dependencies import get_current_user |
| from app.db_models import User |
| from app.config import settings |
|
|
| router = APIRouter(prefix="/auth", tags=["authentication"]) |
|
|
| |
| from slowapi import Limiter |
| from slowapi.util import get_remote_address |
|
|
| limiter = Limiter(key_func=get_remote_address) |
|
|
|
|
| class UserInfo(BaseModel): |
| id: int |
| email: str |
| username: str |
| avatar_url: Optional[str] |
| role: str |
|
|
| class Config: |
| from_attributes = True |
|
|
|
|
| @router.get("/me", response_model=UserInfo) |
| @limiter.limit("60/minute") |
| async def get_current_user_info( |
| request: Request, |
| current_user: Optional[User] = Depends(get_current_user), |
| ): |
| if not current_user: |
| raise HTTPException(status_code=401, detail="Not authenticated") |
|
|
| return UserInfo( |
| id=current_user.id, |
| email=current_user.email, |
| username=current_user.username, |
| avatar_url=current_user.avatar_url, |
| role=current_user.role, |
| ) |
|
|
|
|
| @router.get("/github") |
| @limiter.limit("10/minute") |
| async def github_login(request: Request): |
| return RedirectResponse( |
| url=f"https://github.com/login/oauth/authorize?client_id={settings.GITHUB_CLIENT_ID}&scope=user:email" |
| ) |
|
|
|
|
| @router.get("/github/callback") |
| @limiter.limit("20/minute") |
| async def github_callback( |
| request: Request, |
| code: str, |
| response: Response, |
| session: AsyncSession = Depends(get_db), |
| ): |
| try: |
| user, token = await oauth_handler.github_callback(code, session) |
|
|
| response = RedirectResponse(url=f"{settings.FRONTEND_URL}/dashboard") |
| response.set_cookie( |
| key="access_token", |
| value=token, |
| httponly=True, |
| secure=True if settings.FRONTEND_URL.startswith("https") else False, |
| samesite="lax", |
| max_age=60 * 60 * 24 * 7, |
| ) |
|
|
| return response |
|
|
| except Exception as e: |
| return RedirectResponse(url=f"{settings.FRONTEND_URL}/login?error={str(e)}") |
|
|
|
|
| @router.get("/google") |
| @limiter.limit("10/minute") |
| async def google_login(request: Request): |
| google_client_id = settings.GOOGLE_CLIENT_ID |
| redirect_uri = f"{settings.API_URL}/auth/google/callback" |
|
|
| return RedirectResponse( |
| url=f"https://accounts.google.com/o/oauth2/v2/auth?" |
| f"client_id={google_client_id}&" |
| f"redirect_uri={redirect_uri}&" |
| f"response_type=code&" |
| f"scope=openid email profile" |
| ) |
|
|
|
|
| @router.get("/google/callback") |
| @limiter.limit("20/minute") |
| async def google_callback( |
| request: Request, |
| code: str, |
| response: Response, |
| session: AsyncSession = Depends(get_db), |
| ): |
| try: |
| redirect_uri = f"{settings.API_URL}/auth/google/callback" |
| user, token = await oauth_handler.google_callback(code, redirect_uri, session) |
|
|
| response = RedirectResponse(url=f"{settings.FRONTEND_URL}/dashboard") |
| response.set_cookie( |
| key="access_token", |
| value=token, |
| httponly=True, |
| secure=True if settings.FRONTEND_URL.startswith("https") else False, |
| samesite="lax", |
| max_age=60 * 60 * 24 * 7, |
| ) |
|
|
| return response |
|
|
| except Exception as e: |
| return RedirectResponse(url=f"{settings.FRONTEND_URL}/login?error={str(e)}") |
|
|
|
|
| @router.post("/logout") |
| @limiter.limit("30/minute") |
| async def logout(request: Request, response: Response): |
| response = Response(content={"message": "Logged out successfully"}) |
| response.delete_cookie(key="access_token") |
| return response |
|
|