""" Shared FastAPI dependencies: current‑user injection, role checks. """ from typing import Annotated, List from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer from jose import JWTError, jwt from sqlalchemy.orm import Session from app.config import settings from app.database import get_db from app.models.user import User oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login") def get_current_user( token: Annotated[str, Depends(oauth2_scheme)], db: Session = Depends(get_db), ) -> User: credentials_exc = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) try: payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]) user_id: int | None = payload.get("sub") if user_id is None: raise credentials_exc except JWTError: raise credentials_exc user = db.query(User).filter(User.id == int(user_id)).first() if user is None or not user.is_active: raise credentials_exc return user def require_roles(allowed: List[str]): """Return a dependency that ensures the user has one of the allowed roles.""" def _check(current_user: User = Depends(get_current_user)) -> User: if current_user.role not in allowed: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Insufficient permissions") return current_user return _check CurrentUser = Annotated[User, Depends(get_current_user)] AdminUser = Annotated[User, Depends(require_roles(["admin"]))] InstructorUser = Annotated[User, Depends(require_roles(["admin", "instructor"]))] StudentUser = Annotated[User, Depends(require_roles(["student"]))]