I have table “categories” with subcategories. Subcategories are determined by “parent_id”.
I want to find a way to return an object from FastAPI’s endpoint like this:
[
{
"name": "Category",
"id": "599a5906-1ba1-45ca-9af9-bac2e1513519",
"created_at": "2024-07-17T21:25:24.590781",
"child": [
{
"name": "SubCategory",
"id": "bc540161-e390-4982-87a8-f8eff47204b6",
"created_at": "2024-07-17T21:25:43.307017",
"child": null
}
]
}
]
SQlAlchemy model (here I created a relationship as adviced in this question):
from sqlalchemy import ForeignKey
from sqlalchemy.orm import Mapped, mapped_column, relationship
from database._types import created_at, uuid, uuidpk
from database.engine import Base
class CategoriesModel(Base):
__tablename__ = "categories"
id: Mapped[uuidpk]
parent_id: Mapped[uuid | None] = mapped_column(ForeignKey("categories.id"))
name: Mapped[str]
created_at: Mapped[created_at]
child: Mapped[list["CategoriesModel"]] = relationship(
"CategoriesModel",
remote_side=[parent_id],
uselist=True
)
Query method from repository:
class SQLAlchemyRepository[T]:
def __init__(self, session: AsyncSession) -> None:
self._model: MappedClassProtocol = get_args(self.__orig_bases__[0])[0]
self._session = session
class CategoriesRepository(SQLAlchemyRepository[CategoriesModel]):
async def get_many(self) -> list[Row]:
stmt = (
select(self._model.id, self._model.name, self._model.created_at)
.order_by(self._model.name)
)
res = await self._session.execute(stmt)
return res.all()
Pydantic models:
from datetime import datetime
from _types import uuid
class SQLAlchemyMappedModel(BaseModel):
class Config:
from_attributes = True
class CategorySchema(SQLAlchemyMappedModel):
name: str
class CategoryOut(CategorySchema):
id: uuid
created_at: datetime
child: list["CategorySchema"] = []
Endpoint:
@router.get("/", summary="Categories")
async def get_categories(
uow: Annotated[IUnitOfWork, Depends(get_uow)],
) -> list[CategoryOut]:
async with uow:
res = await uow.categories.get_many()
await uow.commit()
return res
SQLAlchemy doesn’t return “child”. So the result is different from expected.
[
{
"name": "Category",
"id": "599a5906-1ba1-45ca-9af9-bac2e1513519",
"created_at": "2024-07-17T21:25:24.590781",
"child": []
},
{
"name": "SubCategory",
"id": "bc540161-e390-4982-87a8-f8eff47204b6",
"created_at": "2024-07-17T21:25:43.307017",
"child": []
}
]
I’ve decided not to include code unrelated to the question (FastAPI dependencies, UOW), but if a minimally reproducible example is needed, I can do that.
The problem was in the incorrect query. It should be like this:
from sqlalchemy.orm import joinedload
class CategoriesRepository(SQLAlchemyRepository[CategoriesModel]):
async def get_many(self) -> list[Row]:
stmt = (
select(self._model)
.options(joinedload(self._model.child))
.order_by(self._model.name)
)
res = await self._session.execute(stmt)
return res.unique().scalars().all()
In this way ORM generate a correct SQL query (print(stmt)):
SELECT categories.id, categories.parent_id, categories.name, categories.created_at, categories_1.id AS id_1, categories_1.parent_id AS parent_id_1, categories_1.name AS name_1, categories_1.created_at AS created_at_1
FROM categories
LEFT OUTER JOIN categories AS categories_1
ON categories.id = categories_1.parent_id
ORDER BY categories.name
Some examples about sync usage cases could be found in the official docs. But I couldn’t find any about asynchronous use.
Also I made a mistake in pydantic model:
class CategoryOut(CategorySchema):
id: uuid
created_at: datetime
child: list["CategorySchema"] = []
child field should be annotated by its class:
child: list["CategoryOut"] = []
3