Spaces:
Running
Running
| 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 | |