123456789101112131415161718192021222324252627282930313233343536373839404142434445 |
- import json
- from DCbackend.utils.common import success, fail
- from backend.models import AdminRole, UserOperation, Admin, Operation
- from django.db.models import F, Case, When, IntegerField, Exists, OuterRef
- from django.core.cache import cache
- class UserOperationService:
- @staticmethod
- def verify_token(token):
- admin_id = cache.get(f"admin_token:{token}")
- if admin_id:
- return Admin.objects.filter(id=admin_id).first()
- return None
- @staticmethod
- def selectUserOperation(request):
- data = UserOperationService.verify_token(request.POST.get("token"))
- if not data:
- return fail("token过期请刷新后重试")
- role_id = data.role_id
- operations = Operation.objects.all()
- # 创建一个子查询,检查每个 Operation 是否存在于 UserOperation 中
- user_operation_exists = UserOperation.objects.filter(
- role_id=role_id,
- operation_id=OuterRef('pk')
- )
- # 使用子查询添加 type 字段
- operations = operations.annotate(
- type=Case(
- When(Exists(user_operation_exists), then=1),
- default=0,
- output_field=IntegerField()
- )
- )
- # 将查询结果转换为列表
- operation_list = list(operations.values('content_type_id', 'name', 'type')) # 添加其他需要的字段
- return success(operation_list)
|