from typing import Union, Optional from sqlalchemy.ext.asyncio import AsyncSession from dto.catalogue import CatalogueType, CatalogueView from misc.mapper import parse_cat_id, serialize_cat_id from misc.texts import get_breadcrumps_path from repositories import ProductRepository, CategoriesRepository from config import PAGE_SIZE class CatalogueService: def __init__( self, product_repo: ProductRepository, categories_repo: CategoriesRepository ) -> None: self.product_repo = product_repo self.categories_repo = categories_repo async def build_category_view( self, session: AsyncSession, cat_id: Optional[Union[int, str]], page: int = 0 ) -> CatalogueView: cat_id = parse_cat_id(cat_id) view = CatalogueView(page=page) children = await self.categories_repo.get_categories_by_parent_id( session, cat_id ) category = await self.categories_repo.get_category_by_id(session, cat_id) if not category: # Root category view.category = "root" view.show_menu = True else: # Subcategories view.category = cat_id view.parent_id = serialize_cat_id(category.parent_id) if not children: # Products products = await self.product_repo.get_product_by_category( session, cat_id, limit=PAGE_SIZE + 1, offset=view.page * PAGE_SIZE ) if products: view.view_type = CatalogueType.PRODUCTS view.products = products[:PAGE_SIZE] view.has_next = len(products) > PAGE_SIZE return view view.view_type = CatalogueType.EMPTY return view view.view_type = CatalogueType.CATEGORIES view.children = children return view async def get_path(self, session: AsyncSession, cat_id: Union[int, str]): category = int(cat_id) if isinstance(cat_id, int) or cat_id.isdigit() else None products = await self.categories_repo.get_category_path(session, category) return get_breadcrumps_path(products)