Upload app/core/deps.py

app/core/deps.py

@@ -0,0 +1,68 @@
+"""FastAPI dependency functions for auth and role checks."""
+
+from __future__ import annotations
+
+from fastapi import Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.db import get_db
+from app.core.security import decode_access_token
+from app.models.user import User, UserRole
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")
+
+
+async def get_current_user(
+    token: str = Depends(oauth2_scheme),
+    db: AsyncSession = Depends(get_db),
+) -> User:
+    """Resolve the current authenticated user from the JWT in the Authorization header.
+
+    Raises 401 on invalid/expired token, missing user, or soft-deleted user.
+    """
+    credentials_exc = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+
+    payload = decode_access_token(token)
+    if payload is None:
+        raise credentials_exc
+
+    sub = payload.get("sub")
+    if sub is None:
+        raise credentials_exc
+
+    try:
+        user_id = int(sub)
+    except (ValueError, TypeError):
+        raise credentials_exc from None
+
+    # Load user fresh from DB to honor soft-delete and role changes
+    result = await db.execute(
+        select(User).where(User.id == user_id, User.deleted_at.is_(None))
+    )
+    user = result.scalar_one_or_none()
+
+    if user is None:
+        raise credentials_exc
+
+    return user
+
+
+async def get_current_admin_user(
+    user: User = Depends(get_current_user),
+) -> User:
+    """Require the current user to have the admin role."""
+    user_role = (
+        user.role.value if hasattr(user.role, "value") else str(user.role)
+    )
+    if user_role != UserRole.admin.value:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Admin role required",
+        )
+    return user