如何使用Pydantic标准化API返回的不同格式同类型信息导入?
问题描述
我从Web API请求获取的数据可能返回两种不同的字典格式:
# Wh版本格式 [{'energy_wh': int, 'month': int, 'year': int}] # kWh版本格式 [{'energy': float, 'month': int, 'year': int}]
只能通过检查对应键的存在来判断格式。我希望创建一个包含以下属性的类:
class EmeterMonth: year: int month: int energy: float # 单位kWh energy_wh: float # 单位Wh
需要将两种格式的字典数组转换为EmeterMonth类的实例数组,转换规则如下:
- Wh版本:
energy_wh = float(energy_wh),energy = energy_wh / 1000 - kWh版本:
energy_wh = energy * 1000,energy = energy
要求代码清晰易懂,且能支持后续新增类似格式。目前我通过重写__getitem__实现转换,但方式不够清晰粗糙。实际还有更多需转换单位的数据成员,此处做了简化。有人建议用Pydantic声明格式,但不确定如何处理不同格式,使用Python 3.11。
解决方案
方案一:普通Python类实现(高扩展性)
用类方法拆分不同格式的转换逻辑,预留扩展接口,后续新增格式只需添加对应方法即可:
from typing import List, Dict, Any class EmeterMonth: def __init__(self, year: int, month: int, energy: float, energy_wh: float): self.year = year self.month = month self.energy = energy self.energy_wh = energy_wh @classmethod def from_wh_dict(cls, data: Dict[str, Any]) -> 'EmeterMonth': """从Wh格式字典创建实例""" energy_wh = float(data['energy_wh']) return cls( year=data['year'], month=data['month'], energy=energy_wh / 1000, energy_wh=energy_wh ) @classmethod def from_kwh_dict(cls, data: Dict[str, Any]) -> 'EmeterMonth': """从kWh格式字典创建实例""" energy = data['energy'] return cls( year=data['year'], month=data['month'], energy=energy, energy_wh=energy * 1000 ) @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'EmeterMonth': """自动识别格式并创建实例""" if 'energy_wh' in data: return cls.from_wh_dict(data) elif 'energy' in data: return cls.from_kwh_dict(data) else: raise ValueError("无法识别的字典格式:缺少energy或energy_wh键") # 批量转换工具函数 def convert_to_emeter_month_list(data_list: List[Dict[str, Any]]) -> List[EmeterMonth]: return [EmeterMonth.from_dict(item) for item in data_list]
优势
- 每个格式对应独立的类方法,逻辑清晰,新增格式只需添加
from_xxx_dict方法并修改from_dict的判断逻辑 - 初始化逻辑集中在
__init__,保证所有实例的属性一致性 - 批量转换函数直观易用
方案二:Pydantic实现(类型安全+自动验证)
利用Pydantic的模型验证和预处理能力,既能处理多格式转换,又能自动校验API返回的数据合法性(Python 3.11推荐使用Pydantic v2)。
先安装依赖:
pip install pydantic
简洁实现版
直接在目标模型中添加预处理验证器,统一转换逻辑:
from typing import List from pydantic import BaseModel, model_validator class EmeterMonth(BaseModel): year: int month: int energy: float # 单位kWh energy_wh: float # 单位Wh @model_validator(mode='before') def parse_raw_data(cls, data: dict) -> dict: """预处理原始数据,统一转换为目标格式""" if 'energy_wh' in data: energy_wh = float(data['energy_wh']) return { 'year': data['year'], 'month': data['month'], 'energy': energy_wh / 1000, 'energy_wh': energy_wh } elif 'energy' in data: energy = data['energy'] return { 'year': data['year'], 'month': data['month'], 'energy': energy, 'energy_wh': energy * 1000 } else: raise ValueError("无法识别的字典格式:缺少energy或energy_wh键") # 批量转换直接调用模型的验证方法 def convert_to_emeter_month_list(data_list: List[dict]) -> List[EmeterMonth]: return EmeterMonth.model_validate(data_list)
拆分模型版(更适合复杂场景)
如果后续格式差异大,可以拆分原始格式模型和目标模型,逻辑更清晰:
from typing import List, Union from pydantic import BaseModel # 定义API返回的两种原始格式模型 class EmeterWhRaw(BaseModel): energy_wh: int month: int year: int class EmeterKwhRaw(BaseModel): energy: float month: int year: int # 目标模型 class EmeterMonth(BaseModel): year: int month: int energy: float # 单位kWh energy_wh: float # 单位Wh @classmethod def from_wh_raw(cls, raw_model: EmeterWhRaw) -> 'EmeterMonth': energy_wh = float(raw_model.energy_wh) return cls( year=raw_model.year, month=raw_model.month, energy=energy_wh / 1000, energy_wh=energy_wh ) @classmethod def from_kwh_raw(cls, raw_model: EmeterKwhRaw) -> 'EmeterMonth': return cls( year=raw_model.year, month=raw_model.month, energy=raw_model.energy, energy_wh=raw_model.energy * 1000 ) @classmethod def from_raw_dict(cls, data: dict) -> 'EmeterMonth': """自动识别原始格式并转换""" try: return cls.from_wh_raw(EmeterWhRaw(**data)) except: try: return cls.from_kwh_raw(EmeterKwhRaw(**data)) except: raise ValueError("无法识别的字典格式") # 批量转换工具函数 def convert_to_emeter_month_list(data_list: List[dict]) -> List[EmeterMonth]: return [EmeterMonth.from_raw_dict(item) for item in data_list]
优势
- 自带数据验证,比如API返回的
year不是整数时会自动抛出错误,避免脏数据 - 类型提示完善,IDE能提供精准的代码补全
- 复杂场景下拆分模型的方式更易维护,新增格式只需添加对应的原始模型和转换方法
内容的提问来源于stack exchange,提问作者Jules
相关产品推荐
相关产品推荐

