Spaces:
Running
Running
File size: 1,328 Bytes
29cbab9 | 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 | from typing import Annotated
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlmodel import Session, select
from app.core.config import settings
from app.core.database import get_session
from app.core.security import decode_access_token
from app.models.user import User
reusable_oauth2 = OAuth2PasswordBearer(
tokenUrl=f"/auth/login"
)
def get_current_user(
session: Annotated[Session, Depends(get_session)],
token: Annotated[str, Depends(reusable_oauth2)]
) -> User:
"""
FastAPI dependency to get the current authenticated user.
"""
payload = decode_access_token(token)
user_id = payload.get("sub")
if not user_id:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
user = session.get(User, user_id)
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found",
)
if not user.is_active:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Inactive user",
)
return user
|