update. 初步完成用户管理接口

users/users.py

@@ -0,0 +1,94 @@
+from typing import List
+
+from django.db.models import Prefetch
+from django.shortcuts import get_object_or_404
+from ninja import Router
+from ninja.pagination import paginate, PageNumberPagination
+
+from users.models import User, Role
+from users.schema import UserUpdate, UserOut, UserCreate
+
+router = Router(tags=['users'])
+
+
+@router.get("/", response=List[UserOut], summary="获取用户列表")
+@paginate(PageNumberPagination, page_size=20)
+def list_users(request):
+    # if not has_permission(request.auth.id, "user:list"):
+    #     raise HttpError(403, "权限不足")
+    return User.objects.all()
+
+
+@router.get("/", response=dict)
+def list_users(request, page: int = 1, page_size: int = 10, username: str = None):
+    qs = User.objects.all().order_by("-id")
+
+    if username:
+        qs = qs.filter(username__icontains=username)
+
+    total = qs.count()
+    start = (page - 1) * page_size
+    end = start + page_size
+
+    items = list(qs[start:end].values(
+        "id", "username", "cname", "phone", "email",
+        "avatar", "dept",
+        "is_active", "is_staff", "is_superuser",
+        "last_login", "date_joined"
+    ))
+
+    # roles 需要手动补
+    for i, obj in enumerate(qs[start:end]):
+        items[i]["roles"] = list(obj.roles.values_list("id", flat=True))
+
+    return {"items": items, "count": total}
+
+
+@router.get("/{id}", response=UserOut)
+def get_user(request, id: int):
+    obj = get_object_or_404(User, id=id)
+    data = obj.__dict__
+    data["roles"] = list(obj.roles.values_list("id", flat=True))
+    return data
+
+
+@router.post("/", response=UserOut)
+def create_user(request, data: UserCreate):
+    user = User.objects.create_user(
+        username=data.username,
+        password=data.password
+    )
+
+    for k, v in data.dict(exclude={"password"}).items():
+        setattr(user, k, v)
+
+    user.save()
+
+    if data.roles:
+        user.roles.set(data.roles)
+
+    return user
+
+
+@router.put("/{id}")
+def update_user(request, id: int, data: UserUpdate):
+    user = get_object_or_404(User, id=id)
+
+    if data.password:
+        user.set_password(data.password)
+
+    for k, v in data.dict(exclude_unset=True, exclude={"password", "roles"}).items():
+        setattr(user, k, v)
+
+    user.save()
+
+    if data.roles is not None:
+        user.roles.set(data.roles)
+
+    return {"success": True}
+
+
+@router.delete("/{id}")
+def delete_user(request, id: int):
+    User.objects.filter(id=id).delete()
+    return {"success": True}