(self, instance: Dict)
| 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) |
no test coverage detected