68 lines
2.3 KiB
Python
68 lines
2.3 KiB
Python
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from config import PAGE_SIZE
|
|
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 CategoriesRepository, ProductRepository
|
|
|
|
|
|
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: int | str | None,
|
|
page: int = 0,
|
|
*,
|
|
include_hidden: bool = False,
|
|
) -> 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,
|
|
include_hidden=include_hidden,
|
|
)
|
|
if products:
|
|
view.view_type = CatalogueType.PRODUCTS
|
|
view.products = products[:PAGE_SIZE]
|
|
view.has_next = len(products) > PAGE_SIZE
|
|
view.total = await self.product_repo.count_products_by_category(
|
|
session, cat_id, include_hidden=include_hidden
|
|
)
|
|
|
|
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: 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)
|