(self, instance: dict, is_import_tool, with_valid=True)
| 666 | |
| 667 | @transaction.atomic |
| 668 | def import_(self, instance: dict, is_import_tool, with_valid=True): |
| 669 | if with_valid: |
| 670 | self.is_valid() |
| 671 | ApplicationImportRequest(data=instance).is_valid(raise_exception=True) |
| 672 | user_id = self.data.get("user_id") |
| 673 | workspace_id = self.data.get("workspace_id") |
| 674 | folder_id = instance.get("folder_id") |
| 675 | mk_instance_bytes = instance.get("file").read() |
| 676 | try: |
| 677 | mk_instance = restricted_loads(mk_instance_bytes) |
| 678 | except Exception as e: |
| 679 | raise AppApiException(1001, _("Unsupported file format")) |
| 680 | application = mk_instance.application |
| 681 | tool_list = mk_instance.get_tool_list() |
| 682 | update_tool_map = {} |
| 683 | if len(tool_list) > 0: |
| 684 | tool_id_list = reduce( |
| 685 | lambda x, y: [*x, *y], |
| 686 | [[tool.get("id"), generate_uuid((tool.get("id") + workspace_id or ""))] for tool in tool_list], |
| 687 | [], |
| 688 | ) |
| 689 | # 存在的工具列表 |
| 690 | exits_tool_id_list = [ |
| 691 | str(tool.id) for tool in QuerySet(Tool).filter(id__in=tool_id_list, workspace_id=workspace_id) |
| 692 | ] |
| 693 | # 需要更新的工具集合 |
| 694 | update_tool_map = { |
| 695 | tool.get("id"): generate_uuid((tool.get("id") + workspace_id or "")) |
| 696 | for tool in tool_list |
| 697 | if not exits_tool_id_list.__contains__(tool.get("id")) |
| 698 | } |
| 699 | |
| 700 | tool_list = [ |
| 701 | {**tool, "id": update_tool_map.get(tool.get("id"))} |
| 702 | for tool in tool_list |
| 703 | if not exits_tool_id_list.__contains__(tool.get("id")) |
| 704 | and not exits_tool_id_list.__contains__(generate_uuid((tool.get("id") + workspace_id or ""))) |
| 705 | ] |
| 706 | application_model = self.to_application(application, workspace_id, user_id, update_tool_map, folder_id) |
| 707 | tool_model_list = [self.to_tool(f, workspace_id, user_id) for f in tool_list] |
| 708 | application_model.save() |
| 709 | # 插入授权数据 |
| 710 | UserResourcePermissionSerializer( |
| 711 | data={ |
| 712 | "workspace_id": self.data.get("workspace_id"), |
| 713 | "user_id": self.data.get("user_id"), |
| 714 | "auth_target_type": AuthTargetType.APPLICATION.value, |
| 715 | } |
| 716 | ).auth_resource(str(application_model.id)) |
| 717 | # 插入认证信息 |
| 718 | ApplicationAccessToken( |
| 719 | application_id=application_model.id, access_token=hashlib.md5(str(uuid.uuid7()).encode()).hexdigest()[8:24] |
| 720 | ).save() |
| 721 | if is_import_tool: |
| 722 | if len(tool_model_list) > 0: |
| 723 | QuerySet(Tool).bulk_create(tool_model_list) |
| 724 | QuerySet(ToolWorkflow).bulk_create( |
| 725 | [ |
no test coverage detected