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}