Coverage for polar/models/wallet.py: 90%

28 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-12-05 16:17 +0000

1from enum import StrEnum 1ab

2from typing import TYPE_CHECKING 1ab

3from uuid import UUID 1ab

4 

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) 

14 

15from polar.kit.db.models.base import RecordModel 1ab

16from polar.kit.extensions.sqlalchemy.types import StringEnum 1ab

17 

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 

21 

22 

23class WalletType(StrEnum): 1ab

24 usage = "usage" 1ab

25 billing = "billing" 1ab

26 

27 

28class Wallet(RecordModel): 1ab

29 __tablename__ = "wallets" 1ab

30 __table_args__ = (UniqueConstraint("type", "currency", "customer_id"),) 1ab

31 

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 ) 

39 

40 @declared_attr 1ab

41 def customer(cls) -> Mapped["Customer"]: 1ab

42 return relationship("Customer", lazy="raise_on_sql") 1ab

43 

44 organization: AssociationProxy["Organization"] = association_proxy( 1ab

45 "customer", "organization" 

46 ) 

47 

48 @declared_attr 1ab

49 def balance(cls) -> Mapped[int]: 1ab

50 from .wallet_transaction import WalletTransaction 1ab

51 

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 )