Python中如何将100种异构API的返回值映射到自定义数据模型？

1. 多API异构JSON映射的优化方案 #

你当前的逐API写专属函数的方案能跑通，但100个API的场景下维护成本极高，重复代码量太大，更优的实现是映射规则配置化+通用处理逻辑的架构，把API之间的差异全部抽成配置，不需要为每个API单独写函数。

核心思路：所有第三方API的差异本质是「你的目标字段」到「第三方返回值的取值路径」的映射差异，把这部分差异做成可配置的字典，嵌套结构的取值路径用列表表示即可。
给你基于现有代码改造的可运行示例：

from dataclasses import dataclass
import requests
from typing import Any, Dict, List, Union

@dataclass
class MyCookingData:
    chocolate: str
    cucumber: str = None
    loaf: str = None

# 所有API的映射规则全部存在这个配置里，新增API只需要加配置项
API_MAPPING: Dict[str, Dict[str, Union[str, List[str]]]] = {
    "api_1": {
        "chocolate": "chocolate",
        "cucumber": "cucumber",
        "loaf": "loaf"
    },
    "api_2": {
        "chocolate": "choco",
        "cucumber": "cuc",
        "loaf": "meatloaf"
    },
    # 嵌套结构示例，比如api_3的chocolate字段在data.info.dark_choco下
    "api_3": {
        "chocolate": ["data", "info", "dark_choco"],
        "cucumber": ["data", "spec", "cucumber_length"],
        "loaf": ["data", "weight", "loaf"]
    }
}

sample_obj_1 = {
    "chocolate": "dark",
    "cucumber": "long and green",
    "loaf": "2 pounds",
}

sample_obj_2 = {
    "choco": "milk",
    "cuc": "5",
    "meatloaf": "juicy",
}

def fetch_the_api(url):
    '''Fetch a public url, but return sample objects for the example'''
    # 实际场景放开下面的请求逻辑即可
    # try:
    #     response = requests.get(url, timeout=5)
    #     return response.json()
    # except requests.exceptions.HTTPError as err:
    #     print(err)
    #     return None
    return url # 示例用模拟返回

# 按取值路径从返回数据里拿值的通用工具函数
def get_value_by_path(data: Dict, path: Union[str, List[str]]) -> Any:
    if isinstance(path, str):
        return data.get(path)
    current = data
    for key in path:
        current = current.get(key)
        if current is None:
            break
    return current

# 通用处理函数，所有API共用这一个，不用单独写N个
def process_api(api_id: str, url: str) -> None:
    mapping = API_MAPPING.get(api_id)
    if not mapping:
        raise ValueError(f"未找到API {api_id} 的映射配置")
    api_data = fetch_the_api(url)
    if not api_data:
        return
    # 按配置生成参数字典
    params = {
        field: get_value_by_path(api_data, path)
        for field, path in mapping.items()
    }
    new_data = MyCookingData(**params)
    populate_db(new_data)

def populate_db(data):
    '''Store in db, but print for example'''
    for each_elem in data.__dict__:
        print(each_elem, '=>', data.__dict__[each_elem])

def main():
    print('Get content from API 1:')
    process_api("api_1", sample_obj_1)
    print('************')
    print('Get content from API 2:')
    process_api("api_2", sample_obj_2)
    print('************')

if __name__ == "__main__":
    main()

运行后的输出和你原有代码完全一致，后续新增API只需要在API_MAPPING里加对应的映射规则即可，不需要新增函数。如果遇到个别API有特殊的字段处理逻辑（比如单位转换、字符串拼接），可以在配置里加可选的自定义处理函数字段，通用逻辑里判断存在就调用即可，兼顾通用性和特殊场景需求。

2. SQLAlchemy对比原生PostgreSQL的优势 #

使用SQLAlchemy可以明显简化整个流程：

• 你可以直接把MyCookingData定义为SQLAlchemy的ORM模型，不需要先组装dataclass再转入库逻辑，组装完模型对象直接调用session.add()+session.commit()即可完成入库，不需要手写任何INSERT SQL语句。
• 后续字段调整、索引新增等操作可以直接通过ORM模型+alembic迁移工具完成，不需要修改大量原生SQL代码，维护成本低很多。
• ORM自带类型校验、事务处理、防SQL注入等能力，不需要自己手动实现这些基础逻辑。

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