在我的 fastapi 应用程序中，当我的 SQLAlchemy 模型具有关系时，如何使异步 SQLAlchemy 与 Pydantic 良好配合？

Question

我正在构建一个使用 Async SQLAlchemy 的 fastapi CRUD 应用程序。当然它使用 Pydantic 因为 fastapi 就在图中。这是我的问题的要点

SQLALchemyModels/

foo.py

    class Foo(Base):

        id_: Mapped[int] = mapped_column(Integer, priamry_key=True)
        bar_id: Mapped[int] = mapped_column(Integer, ForeignKey="bar.id_")
        bar: Mapped["Bar"] = relationship("Bar", back_populates="foo")

bar.py

    class Bar(Foo):

        id_: Mapped[int] = mapped_column(Integer, primary_key=True)
        foo_id: Mapped[int] = mapped_column(Integer, ForeignKey="foo.id_")
        foo: Mapped["Bar"] = relationship("Foo", back_populates="bar")

PydanticSchemas/

foo.py

    class Foo(BaseModel):

        id_:int = Field(...)
        bar_id: int = Field(...)
        bar: Bar = Field(None)

bar.py

    class Bar(BaseModel):

        id_:int = Field(...)
        foo_id: int = Field(...)
        foo: Foo = Field(None)

如果我在数据库中查询 Foo SQLAlchemy 映射行，我想使用 Foo Pydantic BaseModel 来验证它。我尝试了以下方法来加载 Foo

中的栏关系

SQLAlchemy selectinload/selectload/subqueryload

由于这些加载策略发出一个查询，该查询仅在调用时加载 bar 对象，因此我尝试创建一个 pydantic 字段验证器来加载 bar。

class Foo(BaseModel):

    id_: int = Field(...)
    bar_id: int = Field(...)
    bar: Bar = Field(None)

    @field_validator("bar", mode="before")
    @classmethod
    def validate_bar(cls, v):
        if isinstance(v, SQLALchemy.orm.Query):
            v.all()
        return v

此验证显然失败，因为我使用的是异步 SQLAlchemy 并且无法在同步上下文中等待 v.all() 调用。

2. SQLAlchemy 加入负载

Joinedload 为连接表中的字段分配创意名称，因此几乎不可能以狡猾的方式验证它们

我现在倾向于分别从我的 SQLAlchemy 模型和 Pydantic 类中删除关系和相应的字段。然后我可以在路径操作函数中加载 Foo 对象，然后使用它的 id 来查询（SELECT IN 查询）bar 对象。这似乎过于复杂。有更好的办法吗？

Answer 1

如果你想在查询 foo 时自动加载 bar，并且不使用惰性 loa，你可以使用：

foo: Mapped["Bar"] = relationship("Foo", back_populates="bar", lazy="selectin")

https://docs.sqlalchemy.org/en/20/orm/queryguide/relationships.html