创建任务流
(
self, used_nodes: List[Dict], target_nodes: List[Dict],
strategy: dict, flow_data: List[dict])
| 857 | self.close() |
| 858 | |
| 859 | def create_flow( |
| 860 | self, used_nodes: List[Dict], target_nodes: List[Dict], |
| 861 | strategy: dict, flow_data: List[dict]) -> Tuple[Optional[Flow], str]: |
| 862 | """创建任务流""" |
| 863 | f = Flow( |
| 864 | server_ids="|{}|".format("|".join([str(item["id"]) for item in target_nodes])), |
| 865 | step_count=len(flow_data), |
| 866 | strategy=strategy, |
| 867 | status="waiting" |
| 868 | ) |
| 869 | f.id = self.Flow.create(f) |
| 870 | if not isinstance(f.id, int): |
| 871 | return None, "创建任务失败:{}".format(f.id) |
| 872 | step_index = 1 |
| 873 | steps = [] |
| 874 | |
| 875 | def clear(): |
| 876 | self.Flow.delete(f.id) |
| 877 | for step in steps: |
| 878 | if isinstance(step, CommandTask): |
| 879 | self.CommandTask.delete(step.id) |
| 880 | elif isinstance(step, TransferTask): |
| 881 | self.TransferTask.delete(step.id) |
| 882 | |
| 883 | for task_data in flow_data: |
| 884 | if task_data["task_type"] == "command": |
| 885 | cmd_task, msg = self.create_cmd_task(f.id, step_index, task_data, target_nodes) |
| 886 | if not cmd_task: |
| 887 | clear() |
| 888 | return None, msg |
| 889 | steps.append(cmd_task) |
| 890 | elif task_data["task_type"] == "file": |
| 891 | transfer_task, msg = self.create_transfer_task(f.id, step_index, task_data, target_nodes, used_nodes) |
| 892 | if not transfer_task: |
| 893 | clear() |
| 894 | return None, msg |
| 895 | steps.append(transfer_task) |
| 896 | step_index += 1 |
| 897 | |
| 898 | return f, "" |
| 899 | |
| 900 | def create_cmd_task(self, |
| 901 | flow_id: int, step_index: int, task_data: dict, |
no test coverage detected