76 lines
2.3 KiB
Python
76 lines
2.3 KiB
Python
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import delete, func, select, update
|
|
from typing import Optional
|
|
|
|
from db.models.orders import OrderItem
|
|
|
|
|
|
class OrderItemRepository:
|
|
async def get_item_by_id(
|
|
self, session: AsyncSession, item_id: int
|
|
) -> Optional[OrderItem]:
|
|
stmt = select(OrderItem).where(OrderItem.id == item_id)
|
|
return await session.scalar(stmt)
|
|
|
|
async def get_items_by_order(
|
|
self, session: AsyncSession, *, order_id: int, limit: int, offset: int = 0
|
|
) -> list[OrderItem]:
|
|
stmt = (
|
|
select(OrderItem)
|
|
.where(OrderItem.order_id == order_id)
|
|
.limit(limit)
|
|
.offset(offset)
|
|
)
|
|
result = await session.scalars(stmt)
|
|
|
|
return list(result)
|
|
|
|
async def get_items_count(self, session: AsyncSession, *, order_id: int) -> int:
|
|
stmt = select(func.count(OrderItem.id)).where(OrderItem.order_id == order_id)
|
|
return await session.scalar(stmt) or 0
|
|
|
|
async def get_item_by_order_and_product(
|
|
self, session: AsyncSession, *, order_id: int, product_id: int
|
|
) -> Optional[OrderItem]:
|
|
stmt = (
|
|
select(OrderItem)
|
|
.where(OrderItem.order_id == order_id)
|
|
.where(OrderItem.product_id == product_id)
|
|
)
|
|
return await session.scalar(stmt)
|
|
|
|
async def update_order_item_quantity(
|
|
self, session: AsyncSession, order_item_id: int, *, quantity: int = 0
|
|
):
|
|
stmt = (
|
|
update(OrderItem.__table__)
|
|
.where(OrderItem.id == order_item_id)
|
|
.values(quantity=quantity)
|
|
)
|
|
|
|
await session.execute(stmt)
|
|
await session.commit()
|
|
|
|
async def create_order_item(
|
|
self,
|
|
session: AsyncSession,
|
|
*,
|
|
order_id: int,
|
|
product_id: int,
|
|
quantity: int = 1,
|
|
) -> OrderItem:
|
|
order_item = OrderItem(
|
|
order_id=order_id, product_id=product_id, quantity=quantity
|
|
)
|
|
|
|
session.add(order_item)
|
|
await session.commit()
|
|
|
|
return order_item
|
|
|
|
async def delete_item(self, session: AsyncSession, item_id: int):
|
|
stmt = delete(OrderItem.__table__).where(OrderItem.id == item_id)
|
|
|
|
await session.execute(stmt)
|
|
await session.commit()
|