"""Admin-portal user management. Every endpoint requires an admin — regular users cannot create, delete, or reset the password of any account (they can only change their own password via ``/auth/change-password``). """ from fastapi import APIRouter, Depends, Request from sqlalchemy import func, select from sqlalchemy.orm import Session as DbSession from .. import auth from ..auth import require_admin from ..database import get_db from ..errors import APIError from ..models import Admin, Session from ..schemas import AdminCreate, AdminOut, ResetPasswordRequest router = APIRouter(prefix="/users", tags=["users"]) @router.get("", response_model=list[AdminOut]) def list_users(_: Admin = Depends(require_admin), db: DbSession = Depends(get_db)): return db.execute(select(Admin).order_by(Admin.id)).scalars().all() @router.post("", response_model=AdminOut, status_code=201) def create_user( payload: AdminCreate, request: Request, admin: Admin = Depends(require_admin), db: DbSession = Depends(get_db), ): exists = db.execute( select(Admin.id).where(Admin.username == payload.username) ).scalar_one_or_none() if exists is not None: raise APIError(status_code=409, detail=f"User '{payload.username}' already exists") user = Admin( username=payload.username, password_hash=auth.hash_password(payload.password), is_admin=payload.is_admin, must_change_password=payload.force_password_change, ) db.add(user) db.commit() db.refresh(user) auth.record_log( db, username=admin.username, action="create_user", detail=f"Created user '{user.username}'" + (" (admin)" if user.is_admin else ""), request=request, ) return user @router.delete("/{user_id}", status_code=204) def delete_user( user_id: int, request: Request, admin: Admin = Depends(require_admin), db: DbSession = Depends(get_db), ): if user_id == admin.id: raise APIError(status_code=400, detail="You cannot delete your own account") user = db.get(Admin, user_id) if user is None: raise APIError(status_code=404, detail=f"User '{user_id}' not found") if user.is_admin: admin_count = db.execute( select(func.count()).select_from(Admin).where(Admin.is_admin.is_(True)) ).scalar_one() if admin_count <= 1: raise APIError(status_code=400, detail="Cannot delete the last administrator") # Drop the user's sessions too so any live token stops working immediately. for s in db.execute(select(Session).where(Session.admin_id == user.id)).scalars().all(): db.delete(s) db.delete(user) db.commit() auth.record_log( db, username=admin.username, action="delete_user", detail=f"Deleted user '{user.username}'", request=request, ) @router.post("/{user_id}/reset-password", status_code=204) def reset_password( user_id: int, payload: ResetPasswordRequest, request: Request, admin: Admin = Depends(require_admin), db: DbSession = Depends(get_db), ): """Admin resets another user's password; they must change it on next login.""" user = db.get(Admin, user_id) if user is None: raise APIError(status_code=404, detail=f"User '{user_id}' not found") user.password_hash = auth.hash_password(payload.new_password) user.must_change_password = payload.force_password_change # Invalidate existing sessions so the old password/session can't be reused. for s in db.execute(select(Session).where(Session.admin_id == user.id)).scalars().all(): db.delete(s) db.commit() auth.record_log( db, username=admin.username, action="reset_password", detail=f"Reset password for '{user.username}'", request=request, )