from typing import Sequence from sqlalchemy import desc, func, select from fooder.domain import Product, UserProductUsage from fooder.repository.expression import fuzzy_match from fooder.repository.base import RepositoryBase, DEFAULT_LIMIT class ProductRepository(RepositoryBase[Product]): async def get_by_id(self, product_id: int) -> Product: return await self._get(Product.id == product_id) async def get_by_barcode(self, barcode: str) -> Product: return await self._get(Product.barcode == barcode) async def list( self, q: str | None = None, offset: int = 0, limit: int | None = DEFAULT_LIMIT, ) -> Sequence[Product]: expressions = (fuzzy_match(Product.name, q),) if q else () return await self._list(*expressions, offset=offset, limit=limit) async def list_for_user( self, user_id: int, q: str | None = None, offset: int = 0, limit: int | None = DEFAULT_LIMIT, ): usage_label = "usage_count" usage_count = func.coalesce(UserProductUsage.count, 0).label(usage_label) expressions = (fuzzy_match(Product.name, q),) if q else () stmt = ( select(Product, usage_count) .outerjoin( UserProductUsage, (Product.id == UserProductUsage.product_id) & (UserProductUsage.user_id == user_id), ) .order_by(desc(usage_label), Product.name) ) return await self._list(*expressions, offset=offset, limit=limit, stmt=stmt)