| 401 | |
| 402 | |
| 403 | class Query(serializers.Serializer): |
| 404 | workspace_id = serializers.CharField(required=False, label=_("Workspace ID")) |
| 405 | user_id = serializers.UUIDField(required=True, label=_("User ID")) |
| 406 | |
| 407 | def get_query_set(self, instance: Dict, workspace_manage: bool, is_x_pack_ee: bool): |
| 408 | folder_query_set = QuerySet(ApplicationFolder) |
| 409 | application_query_set = QuerySet(Application) |
| 410 | workspace_id = self.data.get("workspace_id") |
| 411 | user_id = self.data.get("user_id") |
| 412 | desc = instance.get("desc") |
| 413 | name = instance.get("name") |
| 414 | publish_status = instance.get("publish_status") |
| 415 | create_user = instance.get("create_user") |
| 416 | if publish_status is not None: |
| 417 | is_publish = True if publish_status == "published" else False |
| 418 | application_query_set = application_query_set.filter(is_publish=is_publish) |
| 419 | if workspace_id is not None: |
| 420 | folder_query_set = folder_query_set.filter(workspace_id=workspace_id) |
| 421 | application_query_set = application_query_set.filter(workspace_id=workspace_id) |
| 422 | folder_id = instance.get("folder_id") |
| 423 | if folder_id is not None and folder_id != workspace_id: |
| 424 | folder_query_set = folder_query_set.filter(parent=folder_id) |
| 425 | application_query_set = application_query_set.filter(folder_id=folder_id) |
| 426 | if name is not None: |
| 427 | folder_query_set = folder_query_set.filter(name__contains=name) |
| 428 | application_query_set = application_query_set.filter(name__contains=name) |
| 429 | if desc is not None: |
| 430 | folder_query_set = folder_query_set.filter(desc__contains=desc) |
| 431 | application_query_set = application_query_set.filter(desc__contains=desc) |
| 432 | if create_user is not None: |
| 433 | application_query_set = application_query_set.filter(user_id=create_user) |
| 434 | application_custom_sql_query_set = application_query_set |
| 435 | application_query_set = application_query_set.order_by("-create_time") |
| 436 | |
| 437 | resource_and_folder_query_set = QuerySet(WorkspaceUserResourcePermission).filter( |
| 438 | auth_target_type="APPLICATION", workspace_id=workspace_id, user_id=user_id |
| 439 | ) |
| 440 | |
| 441 | return ( |
| 442 | { |
| 443 | "application_query_set": application_query_set, |
| 444 | "workspace_user_resource_permission_query_set": resource_and_folder_query_set, |
| 445 | } |
| 446 | if (not workspace_manage) |
| 447 | else { |
| 448 | "application_query_set": application_query_set, |
| 449 | "application_custom_sql": application_custom_sql_query_set, |
| 450 | } |
| 451 | ) |
| 452 | |
| 453 | @staticmethod |
| 454 | def is_x_pack_ee(): |
| 455 | workspace_user_role_mapping_model = DatabaseModelManage.get_model("workspace_user_role_mapping") |
| 456 | role_permission_mapping_model = DatabaseModelManage.get_model("role_permission_mapping_model") |
| 457 | return workspace_user_role_mapping_model is not None and role_permission_mapping_model is not None |
| 458 | |
| 459 | def list(self, instance: Dict): |
| 460 | self.is_valid(raise_exception=True) |