"""Users router: self-service profile and admin user management.""" from fastapi import APIRouter, Depends, HTTPException, status from ..dependencies import get_current_user, require_admin from ..models.users import SetRoleRequest, UpdateUserRequest, UserResponse from ..services.users import ( delete_user, get_user, list_users, set_user_role, update_user, ) router = APIRouter(prefix="/users", tags=["users"]) ALLOWED_ROLES = {"user", "admin"} # --------------------------------------------------------------------------- # Self-service # --------------------------------------------------------------------------- @router.get("/me", response_model=UserResponse) async def get_me(current_user: dict = Depends(get_current_user)) -> UserResponse: user = await get_user(current_user["id"]) if user is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found.") return UserResponse(**user) @router.put("/me", response_model=UserResponse) async def update_me( body: UpdateUserRequest, current_user: dict = Depends(get_current_user), ) -> UserResponse: try: user = await update_user(current_user["id"], email=body.email, password=body.password) except ValueError as exc: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=str(exc)) if user is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found.") return UserResponse(**user) # --------------------------------------------------------------------------- # Admin # --------------------------------------------------------------------------- @router.get("", response_model=list[UserResponse]) async def get_all_users(_: dict = Depends(require_admin)) -> list[UserResponse]: users = await list_users() return [UserResponse(**u) for u in users] @router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT) async def remove_user( user_id: str, current_user: dict = Depends(require_admin), ) -> None: if user_id == current_user["id"]: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Cannot delete your own account.", ) await delete_user(user_id) @router.put("/{user_id}/role", response_model=UserResponse) async def change_role( user_id: str, body: SetRoleRequest, _: dict = Depends(require_admin), ) -> UserResponse: if body.role not in ALLOWED_ROLES: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=f"Role must be one of: {', '.join(sorted(ALLOWED_ROLES))}.", ) user = await set_user_role(user_id, body.role) if user is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found.") return UserResponse(**user)