56 lines
1.5 KiB
Python
56 lines
1.5 KiB
Python
from typing import Optional
|
|
|
|
from sqlalchemy import select, update
|
|
from db.models import Invoice
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from db.models.invoices import InvoiceStatus
|
|
|
|
|
|
class InvoiceRepository:
|
|
async def create_invoice(
|
|
self,
|
|
session: AsyncSession,
|
|
*,
|
|
amount: int,
|
|
creator_id: int,
|
|
inline_message_id: Optional[int] = None,
|
|
status: InvoiceStatus = InvoiceStatus.PENDING,
|
|
) -> Invoice:
|
|
invoice = Invoice(
|
|
creator_id=creator_id,
|
|
amount=amount,
|
|
status=status,
|
|
inline_message_id=inline_message_id,
|
|
)
|
|
|
|
session.add(invoice)
|
|
await session.commit()
|
|
return invoice
|
|
|
|
async def get_invoice_by_id(
|
|
self, session: AsyncSession, invoice_id: int
|
|
) -> Optional[Invoice]:
|
|
stmt = select(Invoice).where(Invoice.id == invoice_id)
|
|
return await session.scalar(stmt)
|
|
|
|
async def update_invoice_msg(
|
|
self, session: AsyncSession, *, invoice_id: int, inline_message_id: int
|
|
):
|
|
stmt = (
|
|
update(Invoice.__table__)
|
|
.where(Invoice.id == invoice_id)
|
|
.values(inline_message_id=inline_message_id)
|
|
)
|
|
await session.execute(stmt)
|
|
|
|
async def update_invoice_status(
|
|
self, session: AsyncSession, *, invoice_id: int, status: InvoiceStatus
|
|
):
|
|
stmt = (
|
|
update(Invoice.__table__)
|
|
.where(Invoice.id == invoice_id)
|
|
.values(status=status)
|
|
)
|
|
await session.execute(stmt)
|