(self, current_page: int, page_size: int, user_id: str, with_valid=True)
| 285 | self.get_query_set()] |
| 286 | |
| 287 | def page(self, current_page: int, page_size: int, user_id: str, with_valid=True): |
| 288 | if with_valid: |
| 289 | self.is_valid(raise_exception=True) |
| 290 | result = page_search(current_page, page_size, |
| 291 | self.get_query_set(), |
| 292 | post_records_handler=lambda u: UserInstanceSerializer(u).data) |
| 293 | role_model = DatabaseModelManage.get_model("role_model") |
| 294 | user_role_relation_model = DatabaseModelManage.get_model("workspace_user_role_mapping") |
| 295 | |
| 296 | def _get_user_roles(user_ids, is_admin=True): |
| 297 | workspace_model = DatabaseModelManage.get_model("workspace_model") |
| 298 | if not (role_model and user_role_relation_model and workspace_model): |
| 299 | return {} |
| 300 | |
| 301 | workspace_mapping = {str(workspace_model.id): workspace_model.name for workspace_model in |
| 302 | workspace_model.objects.all()} |
| 303 | |
| 304 | # 获取所有相关角色关系,并预加载角色信息 |
| 305 | user_role_relations = ( |
| 306 | user_role_relation_model.objects |
| 307 | .filter(user_id__in=user_ids) |
| 308 | .select_related('role') |
| 309 | .distinct('user_id', 'role_id', 'workspace_id') # 确保组合唯一性 |
| 310 | ) |
| 311 | |
| 312 | # 构建用户ID到角色名称列表的映射 |
| 313 | user_role_mapping = defaultdict(set) # 使用 set 去重 |
| 314 | # 构建用户ID到角色ID与工作空间ID映射 |
| 315 | user_role_setting_mapping = defaultdict(lambda: defaultdict(list)) |
| 316 | user_role_workspace_mapping = defaultdict(lambda: defaultdict(list)) |
| 317 | |
| 318 | for relation in user_role_relations: |
| 319 | user_id = str(relation.user_id) |
| 320 | role_id = relation.role_id |
| 321 | workspace_id = relation.workspace_id |
| 322 | if not is_admin and relation.role.type == RoleConstants.ADMIN.name: |
| 323 | continue |
| 324 | user_role_mapping[user_id].add(relation.role.role_name) |
| 325 | user_role_setting_mapping[user_id][role_id].append(workspace_id) |
| 326 | user_role_workspace_mapping[user_id][relation.role.role_name].append( |
| 327 | workspace_mapping.get(workspace_id, workspace_id)) |
| 328 | |
| 329 | # 将 set 转换为 list 以符合返回格式 |
| 330 | user_role_mapping = {uid: list(roles) for uid, roles in user_role_mapping.items()} |
| 331 | |
| 332 | # 转换为所需的结构 |
| 333 | result_user_role_setting_mapping = { |
| 334 | user_id: [{"role_id": role_id, "workspace_ids": workspace_ids} |
| 335 | for role_id, workspace_ids in roles.items()] |
| 336 | for user_id, roles in user_role_setting_mapping.items() |
| 337 | } |
| 338 | result_user_role_workspace_mapping = { |
| 339 | user_id: {role_name: workspace_names |
| 340 | for role_name, workspace_names in roles.items()} |
| 341 | for user_id, roles in user_role_workspace_mapping.items() |
| 342 | } |
| 343 | |
| 344 | return user_role_mapping, result_user_role_setting_mapping, result_user_role_workspace_mapping |
no test coverage detected