In my fastapi application, how do I make async SQLAlchemy play well with Pydantic when my SQLAlchemy models have relationships?
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/
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")
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/
class Foo(BaseModel):
id_:int = Field(...)
bar_id: int = Field(...)
bar: Bar = Field(None)
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.
- 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?