| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354 |
- """Build nested category tree from global_tree_category."""
- from __future__ import annotations
- from typing import Any
- from supply_infra.db.models.global_tree_category import GlobalTreeCategory
- from supply_infra.db.repositories.global_tree_category_repo import GlobalTreeCategoryRepository
- from supply_infra.db.session import get_session
- def _normalize_parent_id(parent_id: int | None) -> int | None:
- if parent_id is None or parent_id == 0:
- return None
- return parent_id
- def _build_children_map(
- categories: list[GlobalTreeCategory],
- ) -> dict[int | None, list[GlobalTreeCategory]]:
- children_map: dict[int | None, list[GlobalTreeCategory]] = {}
- for category in categories:
- parent_key = _normalize_parent_id(category.parent_id)
- children_map.setdefault(parent_key, []).append(category)
- for children in children_map.values():
- children.sort(key=lambda c: (c.level or 0, c.id))
- return children_map
- def _to_node(
- category: GlobalTreeCategory,
- children_map: dict[int | None, list[GlobalTreeCategory]],
- ) -> dict[str, Any]:
- children = [
- _to_node(child, children_map) for child in children_map.get(category.id, [])
- ]
- return {
- "id": category.id,
- "name": category.name,
- "level": category.level,
- "description": category.description,
- "children": children,
- }
- def build_category_tree() -> list[dict[str, Any]]:
- """Load all active categories and return nested root nodes."""
- with get_session() as session:
- repo = GlobalTreeCategoryRepository(session)
- categories = repo.list_active_categories()
- # Build JSON while session is open to avoid DetachedInstanceError.
- children_map = _build_children_map(categories)
- return [_to_node(root, children_map) for root in children_map.get(None, [])]
|
