42 lines
1.2 KiB
Python
42 lines
1.2 KiB
Python
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from db.models.invoice import Invoice
|
|
from schemas.invoices import InvoiceStatus
|
|
|
|
|
|
class InvoiceRepository:
|
|
def __init__(self, session: AsyncSession) -> None:
|
|
self.session = session
|
|
|
|
async def get_by_id(self, id: int) -> Invoice | None:
|
|
stmt = select(Invoice).where(Invoice.id == id)
|
|
r = await self.session.execute(stmt)
|
|
|
|
return r.scalar_one_or_none()
|
|
|
|
async def get_by_user_id(self, user_id: int) -> list[Invoice]:
|
|
stmt = select(Invoice).where(Invoice.creator_id == user_id)
|
|
r = await self.session.execute(stmt)
|
|
|
|
return list(r.scalars().all())
|
|
|
|
async def create(self, creator_id: int, amount: int | float, status: InvoiceStatus) -> Invoice:
|
|
obj = Invoice(
|
|
creator_id=creator_id,
|
|
amount=amount,
|
|
status=status,
|
|
)
|
|
|
|
self.session.add(obj)
|
|
await self.session.commit()
|
|
|
|
return obj
|
|
|
|
async def update_status_by_id(self, invoice_id: int, status: InvoiceStatus) -> Invoice | None:
|
|
invoice = await self.get_by_id(invoice_id)
|
|
invoice.status = status
|
|
await self.session.commit()
|
|
|
|
return invoice
|