from fastapi import APIRouter, Depends, HTTPException, status from .models import UserInDb, User, Role, Token, UserPublic from .models import get_current_active_user, authenticate_user, create_access_token , fake_db, get_current_user from datetime import timedelta, datetime, timezone from ..config import ACCESS_TOKEN_EXPIRE_MINUTES from typing import Annotated, Optional from fastapi.security import OAuth2PasswordRequestForm router = APIRouter( prefix="/auth", tags=["auth"], responses={404: {"description": "Not found"}}, dependencies=[], ) @router.get("/me") async def read_users_me( current_user: Annotated[User, Depends(get_current_active_user)], ) -> UserPublic: return current_user def ADMIN(current_user: Annotated[UserInDb, Depends(get_current_user)]): if current_user.role != Role.admin: raise HTTPException(status_code=400, detail="You are not admin") return current_user @router.get('/home') async def home(current_user : Annotated[User, Depends(ADMIN)]): return {"message" : f"Welcome to home page {current_user.username}"} @router.post('/login') async def login_for_access_token( form_data : Annotated[OAuth2PasswordRequestForm, Depends()], ) -> Token: user = authenticate_user(fake_db, form_data.username, form_data.password) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password", headers={"WWW-Authenticate": "Bearer"}, ) access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) access_token = create_access_token( data={"sub": user['username'], "role": user['role'], 'status': user['status']}, expires_delta=access_token_expires ) return Token(access_token=access_token, token_type="bearer")