使用Pydantic解析并展平复杂第三方API JSON数据

核心结论 #

完全可以用Pydantic实现你提到的所有需求——包括字段展平、去重、忽略未映射字段，甚至在解析过程中发起额外API调用获取用户名这类关联数据。

分步实现方案 #

1. 基础字段展平与忽略未映射字段 #

Pydantic的Field、model_validator配合model_config里的extra='ignore'，可以轻松处理字段映射、展平，同时自动忽略未定义的冗余字段。

假设第三方API返回结构示例：

{
  "data": [
    {
      "entity": {
        "links": { "self": "https://api.example.com/items/1" },
        "attributes": { "name": "Test Item", "status": "active" },
        "fields": {
          "Project": { "value": "Project Alpha" },
          "Description": { "value": "Sample desc" },
          "name": { "value": "Duplicate Name" }
        },
        "relationships": {
          "created_by": { "data": { "id": "user_123" } },
          "other_rel": { "data": { "id": "xxx" } }
        }
      }
    }
  ]
}

对应的Pydantic模型实现：

from pydantic import BaseModel, model_validator
from typing import Optional, Dict

# 定义第三方API的原始结构（仅保留需要用到的字段）
class EntityLinks(BaseModel):
    self: str

class EntityAttributes(BaseModel):
    name: str
    status: str

class FieldItem(BaseModel):
    value: str

class RelationshipData(BaseModel):
    id: str

class EntityRelationships(BaseModel):
    created_by: Optional[Dict[str, RelationshipData]] = None

class Entity(BaseModel):
    links: EntityLinks
    attributes: EntityAttributes
    fields: Dict[str, FieldItem]
    relationships: Optional[EntityRelationships] = None

class RawApiResponse(BaseModel):
    data: list[Entity]

# 定义最终的目标数据容器
class ParsedItem(BaseModel):
    link: str
    name: str
    status: str
    project: str
    description: str
    created_by_username: Optional[str] = None

    model_config = {
        "extra": "ignore"  # 自动忽略未映射的字段
    }

    @model_validator(mode='before')
    def parse_raw_entity(cls, values):
        # 1. 将links.self转为顶级link字段
        values['link'] = values['links']['self']
        # 2. 展平attributes层级
        values.update(values['attributes'])
        # 3. 展平fields并去重（attributes字段优先级高于fields）
        for field_name, field_data in values['fields'].items():
            target_key = field_name.lower()
            if target_key not in values:
                values[target_key] = field_data['value']
        # 4. 提取fields中Project的value作为顶级project字段
        values['project'] = values['fields']['Project']['value']
        # 清理中间冗余字段
        del values['links'], values['attributes'], values['fields']
        return values

2. 解析过程中发起额外API调用 #

如果需要根据relationships里的用户ID调用其他接口获取用户名，可以直接在model_validator中加入同步/异步API请求逻辑。

同步请求示例

import requests

@model_validator(mode='before')
def parse_raw_entity(cls, values):
    # 保留前面的展平逻辑...

    # 5. 解析relationships并调用API获取用户名
    if 'relationships' in values and values['relationships'].get('created_by'):
        user_id = values['relationships']['created_by']['data']['id']
        resp = requests.get(f'https://api.example.com/users/{user_id}')
        if resp.ok:
            values['created_by_username'] = resp.json()['username']
    del values['relationships']
    return values

异步请求示例（适配异步项目）

from pydantic import AsyncModel
import aiohttp

class ParsedItem(AsyncModel):
    # 字段定义不变...

    @model_validator(mode='before')
    async def parse_raw_entity(cls, values):
        # 保留前面的展平逻辑...

        # 异步调用用户信息API
        if 'relationships' in values and values['relationships'].get('created_by'):
            user_id = values['relationships']['created_by']['data']['id']
            async with aiohttp.ClientSession() as session:
                async with session.get(f'https://api.example.com/users/{user_id}') as resp:
                    if resp.status == 200:
                        user_data = await resp.json()
                        values['created_by_username'] = user_data['username']
        del values['relationships']
        return values

3. 批量解析data数组 #

拿到原始API响应后，先解析为原始结构模型，再批量转换为目标容器：

# 同步场景
raw_response = requests.get('https://api.example.com/data').json()
parsed_raw = RawApiResponse(**raw_response)
final_items = [ParsedItem(**item.model_dump()) for item in parsed_raw.data]

# 异步场景需用asyncio运行
# import asyncio
# final_items = await asyncio.gather(*[ParsedItem(**item.model_dump()) for item in parsed_raw.data])

关键特性说明 #

• 忽略未映射字段：通过model_config = {"extra": "ignore"}自动丢弃所有未在目标模型中定义的字段。
• 字段去重逻辑：处理fields时优先保留已存在的字段（比如attributes中的name），避免被fields中的重复字段覆盖。
• 灵活的字段转换：model_validator提供了数据验证前的钩子，可自由实现任意字段提取、展平、重命名逻辑。
• 外部API集成：验证器支持同步/异步请求，可在解析流程中无缝获取关联数据。

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