(cls, get)
| 240 | |
| 241 | @classmethod |
| 242 | def get_task_info(cls, get): |
| 243 | task_id = get.get('task_id/d', 0) |
| 244 | if not task_id: |
| 245 | return json_response(status=False, msg="任务ID不能为空") |
| 246 | |
| 247 | e_db = TaskFlowsDB() |
| 248 | task = e_db.CommandTask.find("id = ?", (task_id,)) |
| 249 | if not task: |
| 250 | return json_response(status=False, msg="任务不存在") |
| 251 | |
| 252 | task_dict = task.to_dict() |
| 253 | task_dict["log_list"] = [] |
| 254 | server_cache = {} |
| 255 | log_list = e_db.CommandLog.query("command_task_id = ?", (task_id,)) |
| 256 | for i in log_list: |
| 257 | tmp = i.to_dict() |
| 258 | if i.status != 0: |
| 259 | tmp["log"] = i.get_log() |
| 260 | tmp["server_name"] = cls.get_server_info(i.server_id, server_cache).get("remarks", "") |
| 261 | task_dict["log_list"].append(tmp) |
| 262 | |
| 263 | return json_response(status=True, msg="获取成功", data=task_dict) |
| 264 | |
| 265 | @staticmethod |
| 266 | def delete_task(get): |
nothing calls this directly
no test coverage detected