from typing import Annotated from fastapi import APIRouter, Body, Cookie, Depends, Form, HTTPException, Response from fastapi.security import OAuth2PasswordRequestForm from src.api.dependacies.db_dep import sessionDep from src.api.dependacies.user_dep import ActiveUser, RefreshUser, http_bearer from src.core.settings import settings from src.schemas.auth import Token from src.schemas.users import UserRequestADD from src.services.auth import AuthService from src.services.users import UserService router = APIRouter( prefix=settings.api.v1.auth, tags=["Auth"], dependencies=[Depends(http_bearer)] ) @router.post(path="/signup") async def registration(session: sessionDep, credential: UserRequestADD): auth = await AuthService(session).registration(credential) return auth @router.post(path="/login", response_model=Token) async def login( session: sessionDep, credential: Annotated[OAuth2PasswordRequestForm, Depends()], response: Response, fingerprint: str = Form(min_length=5), ): result = await AuthService(session).login( credential.username, credential.password, fingerprint=fingerprint ) response.set_cookie( key="refresh_token", value=result["refresh_token"], httponly=True, samesite="lax", path=settings.api.v1_login_url, max_age=60 * 60 * 24 * settings.refresh_token.expire_days, ) return result @router.post(path="/refresh", response_model=Token) async def refresh( session: sessionDep, current_user: RefreshUser, response: Response, fingerprint: Annotated[str, Body(embed=True)], refresh_token: Annotated[str | None, Cookie(name="refresh_token")] = None, ): if refresh_token is None: raise HTTPException(status_code=401, detail="No refresh token") result = await AuthService(session).refresh_tokens( refresh_token, current_user, fingerprint ) response.set_cookie( key="refresh_token", value=result["refresh_token"], httponly=True, samesite="lax", path=settings.api.v1_login_url, max_age=60 * 60 * 24 * settings.refresh_token.expire_days, ) return result @router.get("/me") async def get_me(session: sessionDep, user: ActiveUser): cur_user = await UserService(session).get_user_by_filter_or_raise(id=user.id) return cur_user @router.post(path="/logout") async def logout( session: sessionDep, response: Response, refresh_token: Annotated[str | None, Cookie(name="refresh_token")] = None, ): if refresh_token is None: raise HTTPException(status_code=401, detail="No refresh token") await AuthService(session).delete_token(token=refresh_token) response.delete_cookie(key="refresh_token") return {"status": "ok"}