72 lines
2.4 KiB
Python
72 lines
2.4 KiB
Python
"""Simple user administration with two data-access levels."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, HTTPException
|
|
|
|
from app.auth import AdminUser
|
|
from app.contracts import DataPermission, UserCreate, UserUpdate
|
|
from app.db import create_user, get_user_by_id, list_users, update_user
|
|
from app.security import hash_password
|
|
|
|
router = APIRouter()
|
|
|
|
_PERMISSION_ROLES: dict[DataPermission, str] = {
|
|
"data_manage": "admin",
|
|
"data_view": "operator",
|
|
}
|
|
|
|
|
|
def _permission_from_roles(roles: list[str] | None) -> DataPermission:
|
|
return "data_manage" if "admin" in set(roles or []) else "data_view"
|
|
|
|
|
|
def _public_user(row: dict[str, Any]) -> dict[str, Any]:
|
|
result = dict(row)
|
|
result["permission"] = _permission_from_roles(list(result.pop("roles", []) or []))
|
|
return result
|
|
|
|
|
|
@router.get("/users")
|
|
async def _list_users(_user: AdminUser):
|
|
return [_public_user(row) for row in await list_users()]
|
|
|
|
|
|
@router.post("/users")
|
|
async def _create_user(body: UserCreate, _user: AdminUser):
|
|
data = body.model_dump()
|
|
permission = data.pop("permission")
|
|
password = data.pop("password")
|
|
data["username"] = data["username"].strip()
|
|
data["hashed_password"] = hash_password(password)
|
|
try:
|
|
row = await create_user(data, [_PERMISSION_ROLES[permission]])
|
|
except Exception as exc:
|
|
raise HTTPException(400, f"创建用户失败：{str(exc)[:200]}") from exc
|
|
return _public_user(row)
|
|
|
|
|
|
@router.patch("/users/{user_id}")
|
|
async def _update_user(user_id: int, body: UserUpdate, current_user: AdminUser):
|
|
target = await get_user_by_id(user_id)
|
|
if not target:
|
|
raise HTTPException(404, "用户不存在")
|
|
|
|
data = body.model_dump(exclude_unset=True)
|
|
permission = data.pop("permission", None)
|
|
is_current_user = target["username"] == current_user["username"]
|
|
if is_current_user and data.get("status") == "disabled":
|
|
raise HTTPException(400, "不能停用当前登录账号")
|
|
if is_current_user and permission == "data_view":
|
|
raise HTTPException(400, "不能移除当前账号的数据管理权限")
|
|
|
|
roles = [_PERMISSION_ROLES[permission]] if permission else None
|
|
password = data.pop("password", None)
|
|
if password:
|
|
data["hashed_password"] = hash_password(password)
|
|
row = await update_user(user_id, data, roles)
|
|
if not row:
|
|
raise HTTPException(404, "用户不存在")
|
|
return _public_user(row)
|