52 lines
1.4 KiB
Python
52 lines
1.4 KiB
Python
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from db.models.orders import Order, OrderAddon, OrderStatus
|
|
|
|
|
|
class OrderRepository:
|
|
def __init__(self, session: AsyncSession) -> None:
|
|
self.session = session
|
|
|
|
async def create(
|
|
self,
|
|
user_id: int,
|
|
devices: int,
|
|
duration_days: int,
|
|
addons: list[str] | None = None,
|
|
status: OrderStatus = OrderStatus.PENDING,
|
|
) -> Order:
|
|
if not addons:
|
|
addons = []
|
|
|
|
order = Order(
|
|
user_id=user_id,
|
|
devices=devices,
|
|
duration_days=duration_days,
|
|
status=status,
|
|
)
|
|
self.session.add(order)
|
|
await self.session.flush()
|
|
|
|
for addon_id in addons:
|
|
addon = OrderAddon(order_id=order.id, addon_id=addon_id)
|
|
self.session.add(addon)
|
|
|
|
await self.session.commit()
|
|
|
|
return order
|
|
|
|
async def get_order_by_user_id(self, user_id: int) -> list[Order]:
|
|
stmt = select(Order).where(Order.user_id == user_id)
|
|
r = await self.session.execute(stmt)
|
|
|
|
return list(r.scalars().all())
|
|
|
|
async def get_active_by_user_id(self, user_id: int) -> list[Order]:
|
|
stmt = (
|
|
select(Order).where(Order.user_id == user_id).where(Order.status == OrderStatus.PENDING)
|
|
)
|
|
r = await self.session.execute(stmt)
|
|
|
|
return list(r.scalars().all())
|