from uuid import UUID from fastapi import APIRouter, HTTPException from fastapi.params import Depends from sqlalchemy.orm import Session from app.database import get_db from app.schemas.auth_token import ( AccessToken, TokenRefreshRequest, TokenResponse, ) from app.schemas.user import LoginRequest from app.security.jwt_bearer import JWTBearer from app.services import session_service, token_service, user_service from app.util.errors import NotFoundError router = APIRouter(prefix="/auth", tags=["Authentication (v1)"]) @router.post(path="/login", response_model=TokenResponse) async def login( login_request: LoginRequest, db: Session = Depends(get_db) ): """ Login to a existing account. Creates a new session and returns a access and refresh token. """ user = await user_service.validate_login( db=db, login=login_request ) if not user: raise HTTPException(status_code=403, detail="invalid_email_or_password") session = await session_service.create_session(db=db, user=user, useragent="") token, expire = await token_service.create_access_token( user=user, session_id=session.id ) return TokenResponse( access_token=token, refresh_token=session.refresh_token, not_after=expire ) @router.post(path="/logout", response_model=list[None]) async def logout( db: Session = Depends(get_db), token: AccessToken = Depends(JWTBearer()) ): """ Remove the current session based on the access token, effectively invalidating the current refresh token. """ await session_service.remove_session( db=db, id=UUID(token.session) ) return list() @router.post(path="/refresh", response_model=TokenResponse) async def refresh_access_token( token_request: TokenRefreshRequest, db: Session = Depends(get_db), ): """ Use an existing refresh token to generate a new access token and a new refresh token. """ try: session = await session_service.validate_and_rotate_refresh_token( db=db, refresh_token=token_request.refresh_token ) user = await user_service.get_user(db=db, id=session.user_id) token, expire = await token_service.create_access_token( user=user, session_id=session.id ) return TokenResponse( access_token=token, refresh_token=session.refresh_token, not_after=expire ) except NotFoundError: raise HTTPException(status_code=403, detail="invalid_refresh_token")