如何使用Pydantic嵌套模型重塑扁平字典数据?
解决Pydantic嵌套模型转换扁平数据的问题
问题背景
现有扁平结构的数据源,需要通过Pydantic嵌套模型转换成指定的嵌套响应格式:
{"orders": [{"id": 111, "info": {"dt": "2023-05-11"}}, ...]}
但直接使用parse_obj解析扁平数据时,会触发info字段缺失的验证错误;若将OrderInfo设为可空,又会得到info: null的不符合预期的结果。
解决方案
方案一:手动构造嵌套结构
循环处理每个数据项时,手动提取字段构造嵌套的OrderInfo对象,再传入Order模型:
from pydantic import BaseModel from datetime import date from typing import List from fastapi import FastAPI data = [ {'id': 111, 'dt': '2022-01-13', 'quantity': 5}, {'id': 112, 'dt': '2022-01-14', 'quantity': 10} ] class OrderInfo(BaseModel): dt: date class Order(BaseModel): id: int info: OrderInfo class Orders(BaseModel): orders: List[Order] app = FastAPI() @app.get("/", response_model=Orders) async def get_orders(): orders = [] for item in data: order_info = OrderInfo(dt=item['dt']) order = Order(id=item['id'], info=order_info) orders.append(order) return Orders(orders=orders)
这种方式逻辑直观,适合数据结构简单的场景,调试起来也更方便。
方案二:使用根验证器自动转换
给Order模型添加root_validator,在解析前自动从扁平数据中提取字段构造info结构:
from pydantic import BaseModel, root_validator from datetime import date from typing import List, Dict, Any from fastapi import FastAPI data = [ {'id': 111, 'dt': '2022-01-13', 'quantity': 5}, {'id': 112, 'dt': '2022-01-14', 'quantity': 10} ] class OrderInfo(BaseModel): dt: date class Order(BaseModel): id: int info: OrderInfo @root_validator(pre=True) def build_info(cls, values: Dict[str, Any]) -> Dict[str, Any]: if 'dt' in values: # 提取dt字段,构造info的嵌套结构 values['info'] = {'dt': values.pop('dt')} return values class Orders(BaseModel): orders: List[Order] app = FastAPI() @app.get("/", response_model=Orders) async def get_orders(): orders = [Order.parse_obj(item) for item in data] return Orders(orders=orders)
该方案把转换逻辑封装在模型内部,适合需要批量处理大量数据的场景,代码更整洁。
方案三:Pydantic v2+ 用别名生成器映射字段
如果使用Pydantic v2版本,可以通过配置alias_generator实现扁平字段到嵌套结构的自动映射:
from pydantic import BaseModel, ConfigDict from datetime import date from typing import List from fastapi import FastAPI data = [ {'id': 111, 'dt': '2022-01-13', 'quantity': 5}, {'id': 112, 'dt': '2022-01-14', 'quantity': 10} ] class OrderInfo(BaseModel): dt: date class Order(BaseModel): model_config = ConfigDict( alias_generator=lambda field_name: field_name.replace('info_', '') ) id: int info: OrderInfo class Orders(BaseModel): orders: List[Order] app = FastAPI() @app.get("/", response_model=Orders) async def get_orders(): # 将扁平字段dt重命名为info_dt,匹配别名规则 processed_data = [ {'id': item['id'], 'info_dt': item['dt'], 'quantity': item['quantity']} for item in data ] orders = [Order.model_validate(item, by_alias=True) for item in processed_data] return Orders(orders=orders)
这种方式通过配置实现字段映射,扩展性更强,适合复杂嵌套结构的场景。
内容的提问来源于stack exchange,提问作者Alex
相关产品推荐
相关产品推荐

