Coverage for polar/models/wallet.py: 90%
28 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 15:52 +0000
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 15:52 +0000
1from enum import StrEnum 1ab
2from typing import TYPE_CHECKING 1ab
3from uuid import UUID 1ab
5from sqlalchemy import ForeignKey, String, UniqueConstraint, Uuid, func, select 1ab
6from sqlalchemy.ext.associationproxy import AssociationProxy, association_proxy 1ab
7from sqlalchemy.orm import ( 1ab
8 Mapped,
9 column_property,
10 declared_attr,
11 mapped_column,
12 relationship,
13)
15from polar.kit.db.models.base import RecordModel 1ab
16from polar.kit.extensions.sqlalchemy.types import StringEnum 1ab
18if TYPE_CHECKING: 18 ↛ 19line 18 didn't jump to line 19 because the condition on line 18 was never true1ab
19 from .customer import Customer
20 from .organization import Organization
23class WalletType(StrEnum): 1ab
24 usage = "usage" 1ab
25 billing = "billing" 1ab
28class Wallet(RecordModel): 1ab
29 __tablename__ = "wallets" 1ab
30 __table_args__ = (UniqueConstraint("type", "currency", "customer_id"),) 1ab
32 type: Mapped[WalletType] = mapped_column(StringEnum(WalletType), nullable=False) 1ab
33 currency: Mapped[str] = mapped_column(String(3)) 1ab
34 customer_id: Mapped[UUID] = mapped_column( 1ab
35 Uuid,
36 ForeignKey("customers.id", ondelete="cascade"),
37 unique=True,
38 )
40 @declared_attr 1ab
41 def customer(cls) -> Mapped["Customer"]: 1ab
42 return relationship("Customer", lazy="raise_on_sql") 1ab
44 organization: AssociationProxy["Organization"] = association_proxy( 1ab
45 "customer", "organization"
46 )
48 @declared_attr 1ab
49 def balance(cls) -> Mapped[int]: 1ab
50 from .wallet_transaction import WalletTransaction 1ab
52 return column_property( 1ab
53 select(func.coalesce(func.sum(WalletTransaction.amount), 0))
54 .where(WalletTransaction.wallet_id == cls.id)
55 .correlate_except(WalletTransaction)
56 .scalar_subquery()
57 )