from fastapi import Depends, HTTPException, Request, status from jose import JWTError, jwt from sqlalchemy.orm import Session from app.core.config import settings from app.db.session import get_db from app.models.user import User def get_optional_user(request: Request, db: Session = Depends(get_db)) -> User | None: token = request.cookies.get("access_token") if not token: return None try: payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm]) user_id = int(payload.get("sub")) except (JWTError, TypeError, ValueError): return None return db.query(User).filter(User.id == user_id, User.is_active.is_(True)).first() def get_current_user(request: Request, db: Session = Depends(get_db)) -> User: token = request.cookies.get("access_token") if not token: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Nicht angemeldet.") try: payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm]) user_id = int(payload.get("sub")) except (JWTError, TypeError, ValueError) as exc: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Ungültiger Token.") from exc user = db.query(User).filter(User.id == user_id, User.is_active.is_(True)).first() if not user: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Benutzer nicht gefunden.") return user def get_admin_user(current_user: User = Depends(get_current_user)) -> User: if current_user.role != "admin" and not current_user.is_admin: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin-Rechte erforderlich.") return current_user def require_editor_or_admin(current_user: User = Depends(get_current_user)) -> User: if current_user.role not in {"admin", "editor"}: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Editor- oder Admin-Rechte erforderlich.") return current_user def require_reader_or_higher(current_user: User = Depends(get_current_user)) -> User: if current_user.role not in {"admin", "editor", "reader"}: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin-Rechte erforderlich.") return current_user