UserOperationService.py 1.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445
  1. import json
  2. from DCbackend.utils.common import success, fail
  3. from backend.models import AdminRole, UserOperation, Admin, Operation
  4. from django.db.models import F, Case, When, IntegerField, Exists, OuterRef
  5. from django.core.cache import cache
  6. class UserOperationService:
  7. @staticmethod
  8. def verify_token(token):
  9. admin_id = cache.get(f"admin_token:{token}")
  10. if admin_id:
  11. return Admin.objects.filter(id=admin_id).first()
  12. return None
  13. @staticmethod
  14. def selectUserOperation(request):
  15. data = UserOperationService.verify_token(request.POST.get("token"))
  16. if not data:
  17. return fail("token过期请刷新后重试")
  18. role_id = data.role_id
  19. operations = Operation.objects.all()
  20. # 创建一个子查询,检查每个 Operation 是否存在于 UserOperation 中
  21. user_operation_exists = UserOperation.objects.filter(
  22. role_id=role_id,
  23. operation_id=OuterRef('pk')
  24. )
  25. # 使用子查询添加 type 字段
  26. operations = operations.annotate(
  27. type=Case(
  28. When(Exists(user_operation_exists), then=1),
  29. default=0,
  30. output_field=IntegerField()
  31. )
  32. )
  33. # 将查询结果转换为列表
  34. operation_list = list(operations.values('content_type_id', 'name', 'type')) # 添加其他需要的字段
  35. return success(operation_list)