In my fastapi application, how do I make async SQLAlchemy play well with Pydantic when my SQLAlchemy models have relationships?

Question

I am building a fastapi CRUD application that uses Async SQLAlchemy. Of-course it uses Pydantic since fastapi is in the picture. Here's a gist of my problem

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)

If I query for Foo SQLAlchemy mapped row in the database, I want to validate it using Foo Pydantic BaseModel. I have tried the following approaches to load the bar relationship in Foo

SQLAlchemy selectinload/ selectload/subqueryload

Since these loading strategies emit a query which only loads bar object when called, I tried to create a pydantic field validator to load 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

This validation obviously fails since I am using async SQLAlchemy and I cannot await v.all() call in synchronous context.

2. SQLAlchemy joinedload

Joinedload assigns creative names to fields in joined table and so it becomes almost impossible to pydantically validate them

I have now leaning towards removing relationships and corresponding fields from my SQLAlchemy models and Pydantic classes respectively. I can then load Foo object in my path operation function and then use its id to query (SELECT IN query) bar object. This seems overly complicated. Is there a better way?

Answer 1

If you want automatically load bar when you query foo, and don't use lazy loa, you could use:

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

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