"""User management endpoints (admin only).""" from typing import Any, List from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from app import crud, schemas from app.dependencies import get_db, get_current_user, get_current_active_superuser from app.models.user import User router = APIRouter() @router.get("", response_model=List[schemas.User]) def list_users( *, db: Session = Depends(get_db), current_user: User = Depends(get_current_active_superuser), skip: int = 0, limit: int = 100 ) -> Any: """ List all users (admin only). Requires superuser privileges. """ users = crud.user.get_multi(db, skip=skip, limit=limit) return users @router.get("/{user_id}", response_model=schemas.User) def get_user( *, db: Session = Depends(get_db), user_id: str, current_user: User = Depends(get_current_user) ) -> Any: """ Get user by ID. Users can view their own profile, admins can view any profile. """ user = crud.user.get(db, id=user_id) if not user: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found" ) # Check if user is viewing their own profile or is superuser if user.id != current_user.id and not current_user.is_superuser: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions" ) return user @router.put("/{user_id}", response_model=schemas.User) def update_user( *, db: Session = Depends(get_db), user_id: str, user_in: schemas.UserUpdate, current_user: User = Depends(get_current_user) ) -> Any: """ Update user. Users can update their own profile, admins can update any profile. Only admins can change is_superuser and is_active flags. """ user = crud.user.get(db, id=user_id) if not user: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found" ) # Check permissions if user.id != current_user.id and not current_user.is_superuser: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Not enough permissions" ) # Only superusers can change is_superuser and is_active if not current_user.is_superuser: if ( user_in.is_superuser is not None or user_in.is_active is not None or user_in.permissions is not None ): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Only admins can change user status or permissions" ) user = crud.user.update(db, db_obj=user, obj_in=user_in) return user @router.delete("/{user_id}", response_model=schemas.User) def delete_user( *, db: Session = Depends(get_db), user_id: str, current_user: User = Depends(get_current_active_superuser) ) -> Any: """ Delete user (admin only). Requires superuser privileges. Users cannot delete themselves. """ if user_id == current_user.id: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Users cannot delete themselves" ) user = crud.user.get(db, id=user_id) if not user: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found" ) user = crud.user.remove(db, id=user_id) return user @router.post("", response_model=schemas.User, status_code=status.HTTP_201_CREATED) def create_user( *, db: Session = Depends(get_db), user_in: schemas.UserCreate, current_user: User = Depends(get_current_active_superuser) ) -> Any: """ Create new user (admin only). Requires superuser privileges. """ # Check if username already exists user = crud.user.get_by_username(db, username=user_in.username) if user: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Username already registered" ) # Check if email already exists user = crud.user.get_by_email(db, email=user_in.email) if user: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Email already registered" ) user = crud.user.create(db, obj_in=user_in) return user