如何用Python将boto3获取的CloudWatch告警原始返回转JSON并提取指定字段

针对当前获取非结构化接口返回、需要筛选指定字段生成标准JSON的需求，可通过以下改造逻辑实现目标：

核心改造点 #

• 对接口返回的每条告警数据做字段过滤，仅保留业务需要的内容，剔除冗余字段
• 解决boto3返回的datetime时间对象无法被JSON直接序列化的兼容性问题
• 聚合全部分页拉取的结果，统一生成带缩进、支持中文正常显示的标准JSON格式

可直接复用的代码 #

import boto3
import json
from datetime import datetime

# 自定义需要提取的字段，可根据实际需求增减
REQUIRED_FIELDS = [
    "AlarmName",
    "AlarmArn",
    "AlarmDescription",
    "Namespace",
    "MetricName",
    "StateValue",
    "StateUpdatedTimestamp",
    "Period",
    "EvaluationPeriods",
    "Threshold"
]

# 处理JSON序列化时的特殊类型（AWS返回的时间对象为datetime格式）
def handle_non_serializable(obj):
    if isinstance(obj, datetime):
        return obj.isoformat()
    raise TypeError(f"不支持的序列化类型: {type(obj)}")

# 初始化CloudWatch客户端
cloudwatch = boto3.client("cloudwatch")
paginator = cloudwatch.get_paginator("describe_alarms")

collected_alarms = []
# 遍历分页拉取全量INSUFFICIENT_DATA状态的告警
for response in paginator.paginate(StateValue="INSUFFICIENT_DATA"):
    for alarm_info in response["MetricAlarms"]:
        # 仅提取指定字段
        filtered_alarm = {field: alarm_info.get(field) for field in REQUIRED_FIELDS}
        collected_alarms.append(filtered_alarm)

# 生成标准格式化JSON
result_json = json.dumps(
    collected_alarms,
    indent=2,
    ensure_ascii=False,
    default=handle_non_serializable
)

# 控制台输出结果
print(result_json)

# 可选：将结果写入本地JSON文件
# with open("filtered_cloudwatch_alarms.json", "w", encoding="utf-8") as f:
#     f.write(result_json)

常见问题说明 #

• 为什么原代码直接打印response['MetricAlarms']看起来不是JSON：那是Python字典的字符串表示，没有经过JSON序列化，且默认包含了所有接口返回的冗余字段，同时时间类字段不会被转成可跨系统识别的格式
• 字段缺失的处理：当REQUIRED_FIELDS中定义的字段在某条告警记录中不存在时，会自动赋值为None，不会触发代码报错
• 中文乱码问题：配置ensure_ascii=False后，告警描述中的中文内容会正常显示，不会被转成unicode转义字符

内容的提问来源于stack exchange，提问作者Dexter