Spaces:
Running
Running
| # backend/api/dependencies.py | |
| from fastapi import Depends, HTTPException, status | |
| from fastapi.security import OAuth2PasswordBearer | |
| from sqlalchemy.orm import Session | |
| from jose import JWTError, jwt | |
| from models.user import User | |
| from utils.security import SECRET_KEY, ALGORITHM | |
| from config.database import SessionLocal | |
| oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/token") | |
| def get_db(): | |
| db = SessionLocal() | |
| try: | |
| yield db | |
| finally: | |
| db.close() | |
| def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)) -> User: | |
| """Décode le token JWT pour obtenir l'utilisateur actuel.""" | |
| credentials_exception = HTTPException( | |
| status_code=status.HTTP_401_UNAUTHORIZED, | |
| detail="Impossible de valider les identifiants", | |
| headers={"WWW-Authenticate": "Bearer"}, | |
| ) | |
| try: | |
| payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) | |
| email: str = payload.get("sub") | |
| if email is None: | |
| raise credentials_exception | |
| except JWTError: | |
| raise credentials_exception | |
| user = db.query(User).filter(User.email == email).first() | |
| if user is None: | |
| raise credentials_exception | |
| return user |