我使用Pydantic与FastApi输出ORM数据到JSON。我想扁平化和重新映射ORM模型，以消除JSON中不必要的级别。
这里有一个简化的例子来说明这个问题。
original output: {"id": 1, "billing":
[
{"id": 1, "order_id": 1, "first_name": "foo"},
{"id": 2, "order_id": 1, "first_name": "bar"}
]
}
desired output: {"id": 1, "name": ["foo", "bar"]}
如何从嵌套字典映射值到Pydantic模型?通过使用init提供了一个适用于字典的解决方案函数中的Pydantic模型类。下面的例子展示了如何使用字典:
from pydantic import BaseModel
# The following approach works with a dictionary as the input
order_dict = {"id": 1, "billing": {"first_name": "foo"}}
# desired output: {"id": 1, "name": "foo"}
class Order_Model_For_Dict(BaseModel):
id: int
name: str = None
class Config:
orm_mode = True
def __init__(self, **kwargs):
print(
"kwargs for dictionary:", kwargs
) # kwargs for dictionary: {'id': 1, 'billing': {'first_name': 'foo'}}
kwargs["name"] = kwargs["billing"]["first_name"]
super().__init__(**kwargs)
print(Order_Model_For_Dict.parse_obj(order_dict)) # id=1 name='foo'
(这个脚本已经完成了，它应该像' is '一样运行)
但是，在处理ORM对象时，这种方法不起作用。似乎init函数未被调用。下面是一个示例，它不会提供所需的输出。
from pydantic import BaseModel, root_validator
from typing import List
from sqlalchemy.orm import relationship
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.dialects.postgresql import ARRAY
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
from pydantic.utils import GetterDict
class BillingOrm(Base):
__tablename__ = "billing"
id = Column(Integer, primary_key=True, nullable=False)
order_id = Column(ForeignKey("orders.id", ondelete="CASCADE"), nullable=False)
first_name = Column(String(20))
class OrderOrm(Base):
__tablename__ = "orders"
id = Column(Integer, primary_key=True, nullable=False)
billing = relationship("BillingOrm")
class Billing(BaseModel):
id: int
order_id: int
first_name: str
class Config:
orm_mode = True
class Order(BaseModel):
id: int
name: List[str] = None
# billing: List[Billing] # uncomment to verify the relationship is working
class Config:
orm_mode = True
def __init__(self, **kwargs):
# This __init__ function does not run when using from_orm to parse ORM object
print("kwargs for orm:", kwargs)
kwargs["name"] = kwargs["billing"]["first_name"]
super().__init__(**kwargs)
billing_orm_1 = BillingOrm(id=1, order_id=1, first_name="foo")
billing_orm_2 = BillingOrm(id=2, order_id=1, first_name="bar")
order_orm = OrderOrm(id=1)
order_orm.billing.append(billing_orm_1)
order_orm.billing.append(billing_orm_2)
order_model = Order.from_orm(order_orm)
# Output returns 'None' for name instead of ['foo','bar']
print(order_model) # id=1 name=None
(这个脚本已经完成了，它应该像' is '一样运行)
输出返回name=None，而不是期望的名称列表。
在上面的例子中，我使用Order.from_orm来创建Pydantic模型。这种方法似乎与FastApi在指定响应模型时使用的方法相同。所需的解决方案应该支持在FastApi响应模型中使用，如下例所示:
@router.get("/orders", response_model=List[schemas.Order])
async def list_orders(db: Session = Depends(get_db)):
return get_orders(db)
更新:关于matslinh注释尝试验证器，我替换了init函数，但是，我无法修改返回值以包含新属性。我怀疑这个问题是因为它是一个ORM对象而不是一个真正的字典。下面的代码将提取名称并将其打印到所需的列表中。但是，我不知道如何在模型响应中包含这个更新的结果:
@root_validator(pre=True)
def flatten(cls, values):
if isinstance(values, GetterDict):
names = [
billing_entry.first_name for billing_entry in values.get("billing")
]
print(names)
# values["name"] = names # error: 'GetterDict' object does not support item assignment
return values
我还发现了一些关于这个问题的其他讨论，这些讨论使我尝试了这种方法:https://github.com/samuelcolvin/pydantic/issues/717https://gitmemory.com/issue/samuelcolvin/pydantic/821/744047672
如果重写from_orm类方法怎么办?
class Order(BaseModel):
id: int
name: List[str] = None
billing: List[Billing]
class Config:
orm_mode = True
@classmethod
def from_orm(cls, obj: Any) -> 'Order':
# `obj` is the orm model instance
if hasattr(obj, 'billing'):
obj.name = obj.billing.first_name
return super().from_orm(obj)
在使用FastAPI + Pydantic堆栈时，我真的很想念方便的Django REST框架序列化器…因此，我与GetterDict争吵，以允许在Pydantic模型中定义字段getter函数，如下所示:
class User(FromORM):
fullname: str
class Config(FromORM.Config):
getter_dict = FieldGetter.bind(lambda: User)
@staticmethod
def get_fullname(obj: User) -> str:
return f'{obj.firstname} {obj.lastname}'
其中魔术部分FieldGetter被实现为
from typing import Any, Callable, Optional, Type
from types import new_class
from pydantic import BaseModel
from pydantic.utils import GetterDict
class FieldGetter(GetterDict):
model_class_forward_ref: Optional[Callable] = None
model_class: Optional[Type[BaseModel]] = None
def __new__(cls, *args, **kwargs):
inst = super().__new__(cls)
if cls.model_class_forward_ref:
inst.model_class = cls.model_class_forward_ref()
return inst
@classmethod
def bind(cls, model_class_forward_ref: Callable):
sub_class = new_class(f'{cls.__name__}FieldGetter', (cls,))
sub_class.model_class_forward_ref = model_class_forward_ref
return sub_class
def get(self, key: str, default):
if hasattr(self._obj, key):
return super().get(key, default)
getter_fun_name = f'get_{key}'
if not (getter := getattr(self.model_class, getter_fun_name, None)):
raise AttributeError(f'no field getter function found for {key}')
return getter(self._obj)
class FromORM(BaseModel):
class Config:
orm_mode = True
getter_dict = FieldGetter