"""Admin endpoints for user management. Note: self-service updates (own display name, locale, password) belong to `/auth/*`; this blueprint is admin-only. """ from __future__ import annotations import logging import uuid from flask import Blueprint, jsonify, request from pydantic import BaseModel, Field, ValidationError from app.core.auth_decorators import require_auth, require_perm from app.services import users as users_svc bp = Blueprint("users", __name__, url_prefix="/users") log = logging.getLogger("metamorph.api.users") def _serialize(u: users_svc.UserView) -> dict: return { "id": str(u.id), "email": u.email, "display_name": u.display_name, "locale": u.locale, "is_active": u.is_active, "deleted_at": u.deleted_at.isoformat() if u.deleted_at else None, "created_at": u.created_at.isoformat(), "updated_at": u.updated_at.isoformat(), "groups": [{"id": str(gid), "name": name} for gid, name in u.groups], } class UpdateUserPayload(BaseModel): # display_name: omitted = no change, null = clear, str = set. # Tri-state encoded with a `default-unset` sentinel via model_extra. display_name: str | None = None locale: str | None = Field(default=None, pattern=r"^[a-z]{2}$") is_active: bool | None = None model_config = {"extra": "forbid"} class SetGroupsPayload(BaseModel): group_ids: list[uuid.UUID] model_config = {"extra": "forbid"} def _parse_uuid_or_400(raw: str): try: return uuid.UUID(raw) except ValueError: return None @bp.get("/roster") @require_auth @require_perm("user.read", "mission.create", "mission.update") def list_roster(): """Minimal user list for mission member assignment. Returns only `id`, `email`, `display_name` of active, non-deleted users. Accessible to anyone who can create or update a mission — strictly lighter than `GET /users`, which leaks `is_admin` (via groups), `is_active`, and group memberships and is therefore reserved to `user.read`. """ q = request.args.get("q") or None rows = users_svc.list_users(q=q, is_active=True, limit=200, offset=0)[0] # Sort by email for predictable rendering and stable e2e selectors. return jsonify( { "items": [ { "id": str(u.id), "email": u.email, "display_name": u.display_name, } for u in sorted( (u for u in rows if u.deleted_at is None), key=lambda x: x.email, ) ] } ) @bp.get("") @require_auth @require_perm("user.read") def list_users(): q = request.args.get("q") or None is_active_raw = request.args.get("is_active") is_active: bool | None if is_active_raw is None: is_active = None elif is_active_raw.lower() in ("true", "1", "yes"): is_active = True elif is_active_raw.lower() in ("false", "0", "no"): is_active = False else: return jsonify({"error": "invalid_is_active"}), 400 try: limit = int(request.args.get("limit", "50")) offset = int(request.args.get("offset", "0")) except ValueError: return jsonify({"error": "invalid_pagination"}), 400 limit = max(1, min(limit, 200)) offset = max(0, offset) rows, total = users_svc.list_users(q=q, is_active=is_active, limit=limit, offset=offset) return jsonify( { "items": [_serialize(u) for u in rows], "total": total, "limit": limit, "offset": offset, } ) @bp.get("/") @require_auth @require_perm("user.read") def get_user(user_id: str): uid = _parse_uuid_or_400(user_id) if uid is None: return jsonify({"error": "invalid_id"}), 400 try: u = users_svc.get_user(uid) except users_svc.UserNotFound: return jsonify({"error": "not_found"}), 404 return jsonify(_serialize(u)) @bp.patch("/") @require_auth @require_perm("user.update") def update_user(user_id: str): uid = _parse_uuid_or_400(user_id) if uid is None: return jsonify({"error": "invalid_id"}), 400 raw = request.get_json(silent=True) or {} try: payload = UpdateUserPayload.model_validate(raw) except ValidationError as e: return jsonify({"error": "invalid_request", "details": e.errors()}), 400 # Distinguish "key absent" (no change) from "key=null" (clear) for display_name. display_name_unset = "display_name" not in raw try: u = users_svc.update_user( uid, display_name=... if display_name_unset else payload.display_name, locale=payload.locale, is_active=payload.is_active, ) except users_svc.UserNotFound: return jsonify({"error": "not_found"}), 404 except users_svc.LastAdminProtected as e: return jsonify({"error": "last_admin_protected", "message": str(e)}), 409 log.info( "metamorph.user.updated", extra={ "user_id": str(uid), "fields": sorted(raw.keys()), }, ) return jsonify(_serialize(u)) @bp.delete("/") @require_auth @require_perm("user.delete") def soft_delete(user_id: str): uid = _parse_uuid_or_400(user_id) if uid is None: return jsonify({"error": "invalid_id"}), 400 try: users_svc.soft_delete_user(uid) except users_svc.UserNotFound: return jsonify({"error": "not_found"}), 404 except users_svc.LastAdminProtected as e: return jsonify({"error": "last_admin_protected", "message": str(e)}), 409 log.info("metamorph.user.soft_deleted", extra={"user_id": str(uid)}) return jsonify({"ok": True}) @bp.put("//groups") @require_auth @require_perm("user.update") def set_groups(user_id: str): uid = _parse_uuid_or_400(user_id) if uid is None: return jsonify({"error": "invalid_id"}), 400 try: payload = SetGroupsPayload.model_validate(request.get_json(silent=True) or {}) except ValidationError as e: return jsonify({"error": "invalid_request", "details": e.errors()}), 400 try: u = users_svc.set_user_groups(uid, payload.group_ids) except users_svc.UserNotFound: return jsonify({"error": "not_found"}), 404 except users_svc.LastAdminProtected as e: return jsonify({"error": "last_admin_protected", "message": str(e)}), 409 except ValueError as e: return jsonify({"error": "invalid_request", "message": str(e)}), 400 log.info( "metamorph.user.groups_set", extra={"user_id": str(uid), "groups": [str(g) for g in payload.group_ids]}, ) return jsonify(_serialize(u))