from fastapi import APIRouter, Cookie, Depends, Request, Response from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from src.models.configs_read.env import env_settings from src.models.pydantic_models.model import UserOut from src.service.auth.auth import CurrentUserService, auth_service router=APIRouter(prefix="/protected") oauth2_schema=OAuth2PasswordBearer(tokenUrl="/protected/token", refreshUrl="/protected/refresh") def require_permissions(*permissions: str): #permissions check dependency async def checker( token: str = Depends(oauth2_schema), auth: CurrentUserService = Depends(auth_service), #noqa: B008 ) -> UserOut: return UserOut.model_validate(await auth.get_current_user(token, *permissions)) return checker @router.post("/token") async def get_access_token(request: Request,response:Response,auth:CurrentUserService=Depends(auth_service), form_data:OAuth2PasswordRequestForm=Depends())->dict: # noqa: B008 access_token, refresh_token=await auth.login(form_data_email=form_data.username, form_data_password=form_data.password, request=request) response.set_cookie( key="refresh_token", value=refresh_token, httponly=True, secure=True, samesite="strict", max_age=env_settings.REFRESH_TOKEN_EXPIRE_DAYS * 24 * 60 * 60 ) return {"access_token": access_token, "token_type": "bearer"} @router.post("/refresh") async def get_refresh_token(request:Request,response:Response, refresh_token: str = Cookie(), auth:CurrentUserService=Depends(auth_service))->dict: # noqa: B008 access_token, refresh_token= await auth.refresh_token(refresh_token=refresh_token,request=request) response.set_cookie( key="refresh_token", value=refresh_token, httponly=True, secure=True, samesite="strict", max_age=env_settings.REFRESH_TOKEN_EXPIRE_DAYS * 24 * 60 * 60 ) return {"access_token":access_token, "token_type": "bearer"} @router.get("/logout") async def logout(response:Response,refresh_token: str = Cookie(),auth:CurrentUserService=Depends(auth_service),current_user:UserOut=Depends(require_permissions()))->bool: # noqa: B008 response.delete_cookie("refresh_token") return await auth.logout(refresh_token) @router.get("") async def protected(current_user:UserOut=Depends(require_permissions()))->dict: # noqa: B008 return {"protected router": "Hello, this is a protected router"}