FastAPI处理Prometheus告警：Pydantic嵌套字段解析异常求助

问题根源分析 #

你遇到的问题核心有两点：

1. Pydantic类定义错误：重复定义了endsAt字段（同时指定datetime和str类型），导致解析逻辑混乱。
2. Payload格式理解偏差：你提供的Payload展示了扁平化的键（如labels.namespace），但Prometheus Alertmanager发送的标准Webhook Payload是嵌套JSON结构——labels和annotations本身是独立对象，包含对应子字段。若实际Payload确实是扁平化格式，需额外处理结构转换；若为展示格式错误，直接修正Pydantic类即可。

解决方案步骤 #

1. 适配标准嵌套Payload的Pydantic定义 #

如果实际收到的是标准嵌套结构的Payload（如下方正确格式），按以下方式调整Pydantic类：

正确的Prometheus告警Payload格式

{
  "endsAt": "0001-01-01T00:00:00Z",
  "labels": {
    "prometheus": "label/prometheus-stack",
    "namespace": "namespace_here",
    "alertname": "JobFailed",
    "severity": "critical",
    "job_name": "uuid-job",
    "purpose": "purpose_here",
    "team": "team_here"
  },
  "status": "firing",
  "startsAt": "2023-03-13T21:05:48.433Z",
  "annotations": {
    "summary": "job failed",
    "message": "Job XY failed."
  },
  "fingerprint": "1234123412341234",
  "generatorURL": "https://URL.stuff"
}

修正后的Pydantic类（两种可选方式）

• 方式一：用Dict快速实现
  删除重复的endsAt定义，确保annotations和labels为必填字典类型：

  from datetime import datetime
  from typing import Dict
  from pydantic import BaseModel
  
  class Alert(BaseModel):
      fingerprint: str
      status: str
      startsAt: datetime
      endsAt: datetime  # Pydantic会自动解析ISO格式字符串为datetime对象
      annotations: Dict[str, str]
      labels: Dict[str, str]
      generatorURL: str  # 新增Payload中存在但未定义的字段
  

  访问子字段需用字典索引：

  print(alert.annotations["summary"])
  print(alert.labels["alertname"])
  

• 方式二：用嵌套模型支持点语法
  若想通过alert.annotations.summary的点语法访问，需为嵌套字段定义独立模型：

  from datetime import datetime
  from pydantic import BaseModel
  
  class AlertAnnotations(BaseModel):
      summary: str
      message: str
  
  class AlertLabels(BaseModel):
      prometheus: str
      namespace: str
      alertname: str
      severity: str
      job_name: str
      purpose: str
      team: str
  
  class Alert(BaseModel):
      fingerprint: str
      status: str
      startsAt: datetime
      endsAt: datetime
      annotations: AlertAnnotations
      labels: AlertLabels
      generatorURL: str
  

  此时可直接用点语法访问子字段，符合你最初的期望：

  print(alert.annotations.summary)
  print(alert.labels.alertname)
  

2. 适配扁平化格式的Payload #

如果实际收到的是labels.namespace这种带点分隔键的扁平化Payload，需先转换为嵌套结构再解析，可通过自定义依赖项实现：

from fastapi import Depends, Request
from typing import Dict
import re

def flatten_to_nested(data: Dict) -> Dict:
    nested = {}
    for key, value in data.items():
        parts = key.split('.')
        current = nested
        for part in parts[:-1]:
            if part not in current:
                current[part] = {}
            current = current[part]
        current[parts[-1]] = value
    return nested

async def get_nested_alert(request: Request) -> Dict:
    flat_data = await request.json()
    return flatten_to_nested(flat_data)

@api.post("/alerts/jobs", tags=["Alerts"])
async def recieve_failed_job_alert(alert_data: Dict = Depends(get_nested_alert)):
    alert = Alert(**alert_data)
    print(alert.fingerprint)
    print(alert.annotations.summary)
    print(alert.labels.alertname)

额外注意事项 #

• 确保Pydantic版本兼容（代码示例支持v1和v2，v2中部分导入需改为from pydantic.v1 import BaseModel）。
• 若告警存在可选字段（如部分labels非必现），可给模型字段添加Optional类型和默认值，例如namespace: Optional[str] = None。

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