我有一个字符串列表,表示代码需要过滤掉的特定错误代码。
DMS_WARNING_STOP_REASONS = ["Stop Reason RECOVERABLE_ERROR",
"Stop Reason FATAL_ERROR",
"Stop Reason STOPPED_DUE_TO_LOW_MEMORY",
"Stop Reason STOPPED_DUE_TO_LOW_DISK",
"Stop Reason FAILED"
]
api响应中的StopReason
字段有时会将它们嵌套在较长的字符串中。
dict_response = {'name': 'job_name',
'Status': 'Stopped',
'StopReason': "terminated [reptask/replicationtask.c:2868] [1020101]n Stop Reason FATAL_ERROR Error Level FATAL"}
我尝试过使用find
和其他方法,但不知道如何实现以下行为:检查来自DMS_WARNING_STOP_REASONS
的任何值是否存在/是否是StopReason
字段的一部分。
应执行此搜索的行是:if x.get("StopReason") in self.DMS_WARNING_STOP_REASONS
def process_tasks(self, all_tasks: list) -> dict:
result = {}
for env in self.envs:
tasks = [x for x in all_tasks if env in self.__get_task_id(x)]
warnings = [
{"name": self.__get_task_id(x), "Status": x["Status"],"StopReason": x.get("StopReason", "Stop Reason FAILED")}
for x in tasks
if x.get("StopReason") in self.DMS_WARNING_STOP_REASONS
]
working = [...
]
result[env] = {...
}
return result
我不得不重新使用.get()
,因为它在不将密钥插入dict的情况下实时修改dict。
基于@ILS代码的工作解决方案:
if (any(stop_reason in x.get("StopReason", "Stop Reason FAILED")