generated from gilles/template-webapp
51 lines
1.7 KiB
Python
51 lines
1.7 KiB
Python
"""Repository pour les boutiques."""
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from app.models.shop import Shop
|
|
from app.repositories.base import BaseRepository
|
|
|
|
|
|
class ShopRepository(BaseRepository[Shop]):
|
|
"""Repository pour les opérations sur les boutiques."""
|
|
|
|
def __init__(self, db: AsyncSession) -> None:
|
|
super().__init__(Shop, db)
|
|
|
|
async def get_by_name(self, name: str) -> Shop | None:
|
|
result = await self.db.execute(
|
|
select(Shop).where(Shop.name == name)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def get_with_item_count(self, id: int) -> tuple[Shop, int] | None:
|
|
result = await self.db.execute(
|
|
select(Shop).options(selectinload(Shop.items)).where(Shop.id == id)
|
|
)
|
|
shop = result.scalar_one_or_none()
|
|
if shop is None:
|
|
return None
|
|
return shop, len(shop.items)
|
|
|
|
async def get_all_with_item_count(
|
|
self, skip: int = 0, limit: int = 100
|
|
) -> list[tuple[Shop, int]]:
|
|
result = await self.db.execute(
|
|
select(Shop)
|
|
.options(selectinload(Shop.items))
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.order_by(Shop.name)
|
|
)
|
|
shops = result.scalars().all()
|
|
return [(shop, len(shop.items)) for shop in shops]
|
|
|
|
async def name_exists(self, name: str, exclude_id: int | None = None) -> bool:
|
|
query = select(func.count(Shop.id)).where(Shop.name == name)
|
|
if exclude_id is not None:
|
|
query = query.where(Shop.id != exclude_id)
|
|
result = await self.db.execute(query)
|
|
return result.scalar_one() > 0
|