I use fastapi, postgresDB, I need to create an endpoint that returns tree of product categories
class CategoryTree(BaseModel):
id: int
name: str
children: list[CategoryTree] | None = None
class ListCategoryTree(RootModel):
root: list[CategoryTree]
The table contains information about product categories (postgres DB)
Id | name | parent |
---|---|---|
412 | Women | None |
723 | Costumes | 412 |
9421 | Swimwear | 723 |
3 | Men | None |
633 | Costumes | 3 |
111 | Business suit | 633 |
87 | Kids | None |
With sqlalchemy I get all the categories
def rec(cat, all_categories):
if cat.parent is None:
all_categories.append(CategoryTree(
id=cat.id,
name=cat.name,
children=[]
))
else:
for parent_cat in all_categories:
if parent_cat.id == cat.parent:
parent_cat.children.append(cat)
async def get_category_tree() -> ListCategoryTree:
categories = await category_pg_service.get_categories() # request to db
all_categories = []
for cat in categories:
rec(cat, all_categories)
return ListCategoryTree(root=all_categories) # pydantic model (final result)
@router.get('', response_model=ListCategoryTree)
async def category_tree() -> ListCategoryTree:
return await get_category_tree()
I managed to get a list of parents and their children, but the children also have children. I need help with that
The recursive pattern is -
async def get_category_tree(parent=None):
return [
{
'id': cat['id'],
'name': cat['name'],
'children': await get_category_tree(cat['id']), # recur
}
for cat in await get_categories(parent) # db query
]
I don't know about your DB, but suppose we can query records by parent
-
data = [
{ 'id': 412, 'name': 'Women', 'parent': None },
{ 'id': 723, 'name': 'Costumes', 'parent': 412 },
{ 'id': 9421, 'name': 'Swimwear', 'parent': 723 },
{ 'id': 3, 'name': 'Men', 'parent': None },
{ 'id': 633, 'name': 'Costumes', 'parent': 3 },
{ 'id': 111, 'name': 'Business suit', 'parent': 633 },
{ 'id': 87, 'name': 'Kids', 'parent': None },
]
def get_categories(parent=None):
return mock_query(data, parent=parent)
I could now write a mock_query
to provide a runnable demo -
import asyncio
async def mock_query(data, **kwargs):
await asyncio.sleep(0.1) # simulate async
return list(d for d in data if all(k in d and d[k] == v for k, v in kwargs.items()))
async def main():
print(await get_category_tree()) # build tree
asyncio.run(main())
[
{'id': 412, 'name': 'Women', 'children': [
{'id': 723, 'name': 'Costumes', 'children': [
{'id': 9421, 'name': 'Swimwear', 'children': []}
]}
]},
{'id': 3, 'name': 'Men', 'children': [
{'id': 633, 'name': 'Costumes', 'children': [
{'id': 111, 'name': 'Business suit', 'children': []}
]}
]},
{'id': 87, 'name': 'Kids', 'children': []}
]
In your own code, that might look something more like this -
async def get_category_tree(parent=None):
return ListCategoryTree(root=[
CategoryTree(
id=cat['id'],
name=cat['name'],
children=(await get_category_tree(cat['id'])).root,
)
for cat in await pg_service.get_categories(parent=parent)
])
@router.get('', response_model=ListCategoryTree)
async def category_tree() -> ListCategoryTree:
return await get_category_tree()