File size: 3,872 Bytes
2c40514 | 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 125 126 127 128 129 130 131 132 133 134 135 136 | 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"])
# Access limiter from app state via request
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
|