JSON 处理

JSON 是运维场景最常见的数据交换格式：K8s API、云厂商 SDK、监控 Webhook、CI 产物元数据几乎都返回 JSON。Python 标准库 json 即可完成读写与解析。

---

基本读写

import json

# Python 对象 -> JSON 字符串
payload = {
    "alertname": "HighCPU",
    "labels": {"instance": "10.0.1.10", "severity": "warning"},
    "value": 92.5,
}

text = json.dumps(payload, ensure_ascii=False, indent=2)
print(text)

# JSON 字符串 -> Python 对象
obj = json.loads(text)
print(obj["labels"]["instance"])  # 10.0.1.10

函数	作用
json.dumps()	dict/list → 字符串
json.loads()	字符串 → dict/list
json.dump()	写入文件对象
json.load()	从文件对象读取

---

读写文件

import json
from pathlib import Path

config_path = Path("data/api_config.json")

# 写入
config = {
    "gitlab_url": "https://gitlab.example.com",
    "timeout": 30,
    "projects": ["ops/platform", "ops/scripts"],
}
with config_path.open("w", encoding="utf-8") as f:
    json.dump(config, f, ensure_ascii=False, indent=2)

# 读取
with config_path.open(encoding="utf-8") as f:
    loaded = json.load(f)

print(loaded["gitlab_url"], loaded["projects"])

---

模拟 Webhook 告警解析

import json

webhook_body = """
{
  "receiver": "ops-oncall",
  "alerts": [
    {
      "status": "firing",
      "labels": {"alertname": "DiskFull", "host": "db-01"},
      "annotations": {"summary": "磁盘使用率 95%"}
    },
    {
      "status": "resolved",
      "labels": {"alertname": "DiskFull", "host": "db-01"},
      "annotations": {"summary": "已恢复"}
    }
  ]
}
"""

data = json.loads(webhook_body)
firing = [a for a in data["alerts"] if a["status"] == "firing"]

for alert in firing:
    host = alert["labels"]["host"]
    summary = alert["annotations"]["summary"]
    print(f"[FIRING] {host}: {summary}")

---

安全访问嵌套字段

API 结构多变，直接 data["a"]["b"] 容易 KeyError：

def deep_get(obj: dict, *keys, default=None):
    """逐层 .get，任意一层缺失返回 default。"""
    cur = obj
    for k in keys:
        if not isinstance(cur, dict):
            return default
        cur = cur.get(k)
        if cur is None:
            return default
    return cur


response = {"data": {"items": [{"name": "pod-a"}]}}
name = deep_get(response, "data", "items", 0, "name")
missing = deep_get(response, "data", "items", 0, "cpu", default="N/A")
print(name, missing)

---

序列化注意事项

import json
from datetime import datetime

record = {"time": datetime.now(), "ok": True}

# datetime 不能直接 dumps，需 default 或先转字符串
text = json.dumps(
    record,
    default=str,
    ensure_ascii=False,
)
print(text)

# 中文不要用 ascii 转义
print(json.dumps({"msg": "磁盘告警"}, ensure_ascii=False))
# {"msg": "磁盘告警"}

---

实战：合并多个 JSON 配置文件

import json
from pathlib import Path

def load_json(path: Path) -> dict:
    with path.open(encoding="utf-8") as f:
        return json.load(f)

defaults = load_json(Path("data/defaults.json"))  # 假设存在
override = {"timeout": 60, "retry": 5}

merged = {**defaults, **override}  # 后者覆盖前者
print(json.dumps(merged, ensure_ascii=False, indent=2))

---

小结

• 对外输出用 ensure_ascii=False 保留中文
• 读 API 响应用 .get() 或封装 deep_get
• 写配置文件 indent=2 便于人工 diff
• datetime、自定义类需转换后再序列化

下一章学习 YAML，读写 K8s / Ansible / CI 常见配置。