| 558 | |
| 559 | |
| 560 | class ApplicationSerializer(serializers.Serializer): |
| 561 | workspace_id = serializers.CharField(required=True, label=_("workspace id")) |
| 562 | user_id = serializers.UUIDField(required=True, label=_("User ID")) |
| 563 | |
| 564 | @transaction.atomic |
| 565 | def insert(self, instance: Dict): |
| 566 | work_flow_template = instance.get("work_flow_template") |
| 567 | application_type = instance.get("type") |
| 568 | |
| 569 | # 处理工作流模板安装逻辑 |
| 570 | if work_flow_template: |
| 571 | return self.insert_template_workflow(instance) |
| 572 | if "WORK_FLOW" == application_type: |
| 573 | r = self.insert_workflow(instance) |
| 574 | else: |
| 575 | r = self.insert_simple(instance) |
| 576 | UserResourcePermissionSerializer( |
| 577 | data={ |
| 578 | "workspace_id": self.data.get("workspace_id"), |
| 579 | "user_id": self.data.get("user_id"), |
| 580 | "auth_target_type": AuthTargetType.APPLICATION.value, |
| 581 | } |
| 582 | ).auth_resource(str(r.get("id"))) |
| 583 | return r |
| 584 | |
| 585 | def insert_template_workflow(self, instance: Dict): |
| 586 | self.is_valid(raise_exception=True) |
| 587 | work_flow_template = instance.get("work_flow_template") |
| 588 | download_url = work_flow_template.get("downloadUrl") |
| 589 | if not download_url.startswith("https://apps-assets.fit2cloud.com/"): |
| 590 | raise AppApiException(500, _("Illegal download url")) |
| 591 | # 查找匹配的版本名称 |
| 592 | res = requests.get(download_url, timeout=5) |
| 593 | app = ApplicationSerializer( |
| 594 | data={"user_id": self.data.get("user_id"), "workspace_id": self.data.get("workspace_id")} |
| 595 | ).import_( |
| 596 | { |
| 597 | "file": bytes_to_uploaded_file(res.content, "file.mk"), |
| 598 | "folder_id": instance.get("folder_id", instance.get("workspace_id")), |
| 599 | }, |
| 600 | True, |
| 601 | ) |
| 602 | work_flow = app.get("work_flow") |
| 603 | for node in work_flow.get("nodes", []): |
| 604 | if node.get("type") == "base-node": |
| 605 | node_data = node.get("properties").get("node_data") |
| 606 | node_data["name"] = instance.get("name") |
| 607 | node_data["desc"] = instance.get("desc") |
| 608 | QuerySet(Application).filter(id=app.get("id")).update( |
| 609 | name=instance.get("name"), desc=instance.get("desc"), work_flow=work_flow |
| 610 | ) |
| 611 | try: |
| 612 | download_callback_url = work_flow_template.get("downloadCallbackUrl", "") |
| 613 | if not download_callback_url.startswith("https://apps.fit2cloud.com"): |
| 614 | raise AppApiException(500, _("Illegal download callback url")) |
| 615 | requests.get(download_callback_url, timeout=5) |
| 616 | except Exception as e: |
| 617 | maxkb_logger.error(f"callback appstore tool download error: {e}") |
no outgoing calls
no test coverage detected