46 lines
1.5 KiB
Python
46 lines
1.5 KiB
Python
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)
|