from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from ..database import get_db from ..deps import require_admin from ..models import Role, User from ..schemas import UserCreate, UserOut, UserUpdate from ..security import hash_password router = APIRouter(prefix="/users", tags=["users"], dependencies=[Depends(require_admin)]) @router.get("", response_model=list[UserOut]) def list_users(db: Session = Depends(get_db)) -> list[User]: return db.query(User).order_by(User.username).all() @router.post("", response_model=UserOut, status_code=status.HTTP_201_CREATED) def create_user(payload: UserCreate, db: Session = Depends(get_db)) -> User: if db.query(User).filter(User.username == payload.username).first(): raise HTTPException(status.HTTP_409_CONFLICT, "Benutzername existiert bereits") user = User( username=payload.username, password_hash=hash_password(payload.password), role=payload.role, ) db.add(user) db.commit() db.refresh(user) return user @router.patch("/{user_id}", response_model=UserOut) def update_user( user_id: int, payload: UserUpdate, db: Session = Depends(get_db) ) -> User: user = db.get(User, user_id) if user is None: raise HTTPException(status.HTTP_404_NOT_FOUND, "Benutzer nicht gefunden") if payload.password is not None: user.password_hash = hash_password(payload.password) if payload.role is not None: user.role = payload.role db.commit() db.refresh(user) return user @router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT) def delete_user( user_id: int, db: Session = Depends(get_db), current_admin: User = Depends(require_admin), ) -> None: user = db.get(User, user_id) if user is None: raise HTTPException(status.HTTP_404_NOT_FOUND, "Benutzer nicht gefunden") if user.id == current_admin.id: raise HTTPException(status.HTTP_400_BAD_REQUEST, "Du kannst dich nicht selbst löschen") # Verhindere das Löschen des letzten Admins. if user.role == Role.admin: admin_count = db.query(User).filter(User.role == Role.admin).count() if admin_count <= 1: raise HTTPException( status.HTTP_400_BAD_REQUEST, "Der letzte Administrator kann nicht gelöscht werden" ) db.delete(user) db.commit()