I am using fastapi for the first time, and I divided my project into two parts which are admin (used django) and api (fastapi). First I defined my models in django and migrated, and connect fastapi to the existing db.
models.py
<code>class Category(Base):
__tablename__ = "category"
id = Column(BigInteger, primary_key=True)
created_at = Column(DateTime(timezone=True))
updated_at = Column(DateTime(timezone=True))
icon = Column(String(100))
order = Column(Integer)
lft = Column(Integer)
rght = Column(Integer)
tree_id = Column(Integer)
level = Column(Integer)
parent_id = Column(BigInteger, ForeignKey("category.id"))
parent = relationship("Category", remote_side=[id])
translations = relationship("CategoryTranslation", back_populates="category")
class CategoryTranslation(Base):
__tablename__ = "category_translation"
id = Column(BigInteger, primary_key=True)
language_code = Column(String(15))
name = Column(String(255))
slug = Column(String(255))
master_id = Column(BigInteger, ForeignKey("category.id"))
category = relationship("Category", back_populates="translations")
</code>
<code>class Category(Base):
__tablename__ = "category"
id = Column(BigInteger, primary_key=True)
created_at = Column(DateTime(timezone=True))
updated_at = Column(DateTime(timezone=True))
icon = Column(String(100))
order = Column(Integer)
lft = Column(Integer)
rght = Column(Integer)
tree_id = Column(Integer)
level = Column(Integer)
parent_id = Column(BigInteger, ForeignKey("category.id"))
parent = relationship("Category", remote_side=[id])
translations = relationship("CategoryTranslation", back_populates="category")
class CategoryTranslation(Base):
__tablename__ = "category_translation"
id = Column(BigInteger, primary_key=True)
language_code = Column(String(15))
name = Column(String(255))
slug = Column(String(255))
master_id = Column(BigInteger, ForeignKey("category.id"))
category = relationship("Category", back_populates="translations")
</code>
class Category(Base):
__tablename__ = "category"
id = Column(BigInteger, primary_key=True)
created_at = Column(DateTime(timezone=True))
updated_at = Column(DateTime(timezone=True))
icon = Column(String(100))
order = Column(Integer)
lft = Column(Integer)
rght = Column(Integer)
tree_id = Column(Integer)
level = Column(Integer)
parent_id = Column(BigInteger, ForeignKey("category.id"))
parent = relationship("Category", remote_side=[id])
translations = relationship("CategoryTranslation", back_populates="category")
class CategoryTranslation(Base):
__tablename__ = "category_translation"
id = Column(BigInteger, primary_key=True)
language_code = Column(String(15))
name = Column(String(255))
slug = Column(String(255))
master_id = Column(BigInteger, ForeignKey("category.id"))
category = relationship("Category", back_populates="translations")
I’m trying to retrieve nested objects from my database but I’m unsure how to efficiently query and serialize nested objects.
Here’s a router:
<code>@router.get("/")
async def get_categories(db: Session = Depends(get_db),):
data = db.query(categories.Category).options(
joinedload(categories.Category.translations),
joinedload(categories.Category.parent)
).filter(categories.Category.parent_id.is_(None))
return data.all()
</code>
<code>@router.get("/")
async def get_categories(db: Session = Depends(get_db),):
data = db.query(categories.Category).options(
joinedload(categories.Category.translations),
joinedload(categories.Category.parent)
).filter(categories.Category.parent_id.is_(None))
return data.all()
</code>
@router.get("/")
async def get_categories(db: Session = Depends(get_db),):
data = db.query(categories.Category).options(
joinedload(categories.Category.translations),
joinedload(categories.Category.parent)
).filter(categories.Category.parent_id.is_(None))
return data.all()
This returns me ugly response like:
<code>[
{
"created_at": "2024-07-16T09:25:24.155984+00:00",
"order": 0,
"rght": 4,
"tree_id": 1,
"level": 0,
"parent_id": null,
"updated_at": "2024-07-16T09:25:39.114618+00:00",
"id": 1,
"icon": "",
"lft": 1,
"parent": null,
"translations": [
{
"master_id": 1,
"name": "Design",
"slug": "design",
"id": 1,
"language_code": "uz"
},
{
"master_id": 1,
"name": "дизайн",
"slug": "dizain",
"id": 2,
"language_code": "ru"
}
]
}
]
</code>
<code>[
{
"created_at": "2024-07-16T09:25:24.155984+00:00",
"order": 0,
"rght": 4,
"tree_id": 1,
"level": 0,
"parent_id": null,
"updated_at": "2024-07-16T09:25:39.114618+00:00",
"id": 1,
"icon": "",
"lft": 1,
"parent": null,
"translations": [
{
"master_id": 1,
"name": "Design",
"slug": "design",
"id": 1,
"language_code": "uz"
},
{
"master_id": 1,
"name": "дизайн",
"slug": "dizain",
"id": 2,
"language_code": "ru"
}
]
}
]
</code>
[
{
"created_at": "2024-07-16T09:25:24.155984+00:00",
"order": 0,
"rght": 4,
"tree_id": 1,
"level": 0,
"parent_id": null,
"updated_at": "2024-07-16T09:25:39.114618+00:00",
"id": 1,
"icon": "",
"lft": 1,
"parent": null,
"translations": [
{
"master_id": 1,
"name": "Design",
"slug": "design",
"id": 1,
"language_code": "uz"
},
{
"master_id": 1,
"name": "дизайн",
"slug": "dizain",
"id": 2,
"language_code": "ru"
}
]
}
]
I’d like to get response only with few fields which are name, slug, order, icon, language_code, child
How can I modify my router to achieve this?
You need to create corresponding schema and use it as a response model:
<code>class CategoryTranslationOutpuSchema(BaseModel):
model_config = ConfigDict(from_attributes=True)
name: str
slug: str
language_code: str
class CategoryOutputSchema(BaseModel):
model_config = ConfigDict(from_attributes=True)
order: int
icon: str
translations: list[CategoryTranslationOutpuSchema]
@router.get("/", response_model=list[CategoryOutputSchema])
async def get_categories(db: Session = Depends(get_db),):
data = db.query(categories.Category).options(
joinedload(categories.Category.translations),
joinedload(categories.Category.parent)
).filter(categories.Category.parent_id.is_(None))
return data.all()
</code>
<code>class CategoryTranslationOutpuSchema(BaseModel):
model_config = ConfigDict(from_attributes=True)
name: str
slug: str
language_code: str
class CategoryOutputSchema(BaseModel):
model_config = ConfigDict(from_attributes=True)
order: int
icon: str
translations: list[CategoryTranslationOutpuSchema]
@router.get("/", response_model=list[CategoryOutputSchema])
async def get_categories(db: Session = Depends(get_db),):
data = db.query(categories.Category).options(
joinedload(categories.Category.translations),
joinedload(categories.Category.parent)
).filter(categories.Category.parent_id.is_(None))
return data.all()
</code>
class CategoryTranslationOutpuSchema(BaseModel):
model_config = ConfigDict(from_attributes=True)
name: str
slug: str
language_code: str
class CategoryOutputSchema(BaseModel):
model_config = ConfigDict(from_attributes=True)
order: int
icon: str
translations: list[CategoryTranslationOutpuSchema]
@router.get("/", response_model=list[CategoryOutputSchema])
async def get_categories(db: Session = Depends(get_db),):
data = db.query(categories.Category).options(
joinedload(categories.Category.translations),
joinedload(categories.Category.parent)
).filter(categories.Category.parent_id.is_(None))
return data.all()
And, if you need to filter nested data, you can do it this way:
<code> res = [CategoryOutputSchema.model_validate(category) for category in data]
for category in res:
category.translations = [translation for translation in category.translations if translation.language_code == 'en']
return res
</code>
<code> res = [CategoryOutputSchema.model_validate(category) for category in data]
for category in res:
category.translations = [translation for translation in category.translations if translation.language_code == 'en']
return res
</code>
res = [CategoryOutputSchema.model_validate(category) for category in data]
for category in res:
category.translations = [translation for translation in category.translations if translation.language_code == 'en']
return res
Note: FastAPI will re-validate you response. To avoid second validation you can return JSONResponse directly: https://fastapi.tiangolo.com/advanced/response-directly/
2