1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
|
from enum import StrEnum, auto
from pydantic import BaseModel
from sqlalchemy import and_, select
from sqlalchemy.ext.asyncio import AsyncSession
from models import Invoice, Payment, PaymentStatus, User, UserRole
class InvoiceStatus(StrEnum):
PAID = auto()
UNPAID = auto()
class InvoicePayments(BaseModel):
user_status: dict[int, PaymentStatus]
status: InvoiceStatus
async def get_invoice_payments(
session: AsyncSession,
invoice: Invoice,
) -> InvoicePayments:
users = await session.scalars(
select(User).where(
and_(
User.role != UserRole.ADMIN,
User.datetime <= invoice.datetime,
)
)
)
payments = await session.scalars(
select(Payment).where(Payment.invoice_id == invoice.id)
)
user_status = {u.id: PaymentStatus.REJECTED for u in users}
for p in payments:
if (
p.status != PaymentStatus.REJECTED
and user_status[p.user_id] == PaymentStatus.REJECTED
):
user_status[p.user_id] = p.status
status = (
InvoiceStatus.PAID
if all(s == PaymentStatus.ACCEPTED for s in user_status.values())
else InvoiceStatus.UNPAID
)
return InvoicePayments(user_status=user_status, status=status)
async def get_payment_status(
session: AsyncSession,
invoice_id: int,
user_id: int,
) -> PaymentStatus:
payments = await session.scalars(
select(Payment).where(
and_(
Payment.invoice_id == invoice_id,
Payment.user_id == user_id,
)
)
)
for p in payments:
if p.status != PaymentStatus.REJECTED:
return p.status
return PaymentStatus.REJECTED
|
