将 ORM 展平并重新映射到 Pydantic 模型的最佳方法

Question

我使用 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 模型？通过使用Pydantic 模型类中的init函数，提供了适用于字典的解决方案。此示例展示了它如何与字典一起使用：

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'

（此脚本已完成，它应该“按原样”运行）

然而，当使用 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

（此脚本已完成，它应该“按原样”运行）

输出返回 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)

更新：关于尝试验证器的 MatsLindh 评论，我用根验证器替换了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/717 https://gitmemory.com/issue/samuelcolvin/pydantic/821/744047672

Answer 1

如果重写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)