diff --git a/app/dependencies.py b/app/dependencies.py new file mode 100644 index 0000000..7609aff --- /dev/null +++ b/app/dependencies.py @@ -0,0 +1,69 @@ +""" +Shared dependencies for the application. +""" +from typing import Annotated + +from fastapi import Depends, Header, HTTPException, status +from sqlalchemy.orm import Session + +from app import models +from app.database import SessionLocal +from app.services import auth_service + + +def get_db(): + db = SessionLocal() + try: + yield db + finally: + db.close() + + +def get_current_user( + authorization: Annotated[str | None, Header()] = None, + db: Session = Depends(get_db), +) -> models.User: + """Dependency to get current authenticated user from JWT token.""" + if not authorization: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Non authentifié", + ) + + # Extract token from "Bearer " format + parts = authorization.split() + if len(parts) != 2 or parts[0].lower() != "bearer": + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Format de token invalide", + ) + + token = parts[1] + payload = auth_service.decode_token(token) + + if not payload: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Token invalide ou expiré", + ) + + user = auth_service.get_user_by_id(db, payload["user_id"]) + if not user or not user.is_active: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Utilisateur non trouvé ou désactivé", + ) + + return user + + +def get_admin_user( + current_user: models.User = Depends(get_current_user), +) -> models.User: + """Dependency to require admin privileges.""" + if not current_user.is_admin: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Accès administrateur requis", + ) + return current_user diff --git a/app/routers/auth.py b/app/routers/auth.py index 5eb08fd..f9efc42 100644 --- a/app/routers/auth.py +++ b/app/routers/auth.py @@ -11,18 +11,14 @@ from sqlalchemy.orm import Session from app import models from app.database import SessionLocal from app.services import auth_service +from app.dependencies import get_db, get_current_user, get_admin_user logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/auth", tags=["auth"]) -def get_db(): - db = SessionLocal() - try: - yield db - finally: - db.close() + # Pydantic models @@ -64,54 +60,7 @@ class AdminChangePasswordRequest(BaseModel): new_password: str -def get_current_user( - authorization: Annotated[str | None, Header()] = None, - db: Session = Depends(get_db), -) -> models.User: - """Dependency to get current authenticated user from JWT token.""" - if not authorization: - raise HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, - detail="Non authentifié", - ) - # Extract token from "Bearer " format - parts = authorization.split() - if len(parts) != 2 or parts[0].lower() != "bearer": - raise HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, - detail="Format de token invalide", - ) - - token = parts[1] - payload = auth_service.decode_token(token) - - if not payload: - raise HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, - detail="Token invalide ou expiré", - ) - - user = auth_service.get_user_by_id(db, payload["user_id"]) - if not user or not user.is_active: - raise HTTPException( - status_code=status.HTTP_401_UNAUTHORIZED, - detail="Utilisateur non trouvé ou désactivé", - ) - - return user - - -def get_admin_user( - current_user: models.User = Depends(get_current_user), -) -> models.User: - """Dependency to require admin privileges.""" - if not current_user.is_admin: - raise HTTPException( - status_code=status.HTTP_403_FORBIDDEN, - detail="Accès administrateur requis", - ) - return current_user @router.post("/login", response_model=LoginResponse) diff --git a/app/routers/debug.py b/app/routers/debug.py index 9ae797e..07d694b 100644 --- a/app/routers/debug.py +++ b/app/routers/debug.py @@ -8,11 +8,12 @@ from fastapi.responses import FileResponse from pydantic import BaseModel from app.services import auth_service +from app.dependencies import get_admin_user router = APIRouter( prefix="/debug", tags=["debug"], - dependencies=[Depends(auth_service.get_current_active_admin)], + dependencies=[Depends(get_admin_user)], ) logger = logging.getLogger(__name__)