Coverage for polar/models/payout.py: 78%

63 statements  

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

1from datetime import datetime 1ab

2from enum import StrEnum 1ab

3from typing import TYPE_CHECKING 1ab

4from uuid import UUID 1ab

5 

6from sqlalchemy import TIMESTAMP, ForeignKey, String, UniqueConstraint, Uuid 1ab

7from sqlalchemy.orm import Mapped, mapped_column, relationship 1ab

8from sqlalchemy.sql.sqltypes import BigInteger 1ab

9 

10from polar.enums import AccountType 1ab

11from polar.kit.db.models import RecordModel 1ab

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

13 

14if TYPE_CHECKING: 14 ↛ 15line 14 didn't jump to line 15 because the condition on line 14 was never true1ab

15 from .account import Account 

16 from .transaction import Transaction 

17 

18 

19class PayoutStatus(StrEnum): 1ab

20 pending = "pending" 1ab

21 in_transit = "in_transit" 1ab

22 succeeded = "succeeded" 1ab

23 

24 @classmethod 1ab

25 def from_stripe(cls, stripe_status: str) -> "PayoutStatus": 1ab

26 if stripe_status == "in_transit": 

27 return cls.in_transit 

28 if stripe_status == "paid": 

29 return cls.succeeded 

30 return cls.pending 

31 

32 

33class Payout(RecordModel): 1ab

34 __tablename__ = "payouts" 1ab

35 __table_args__ = (UniqueConstraint("account_id", "invoice_number"),) 1ab

36 

37 processor: Mapped[AccountType] = mapped_column( 1ab

38 StringEnum(AccountType), nullable=False 

39 ) 

40 """Payment processor used for this payout.""" 1ab

41 processor_id: Mapped[str | None] = mapped_column( 1ab

42 String, nullable=True, index=True, unique=False 

43 ) 

44 """ID of the payout in the payment processor. Might be `None` if not yet created.""" 1ab

45 status: Mapped[PayoutStatus] = mapped_column( 1ab

46 StringEnum(PayoutStatus), 

47 nullable=False, 

48 index=True, 

49 default=PayoutStatus.pending, 

50 ) 

51 """Status of this payout.""" 1ab

52 paid_at: Mapped[datetime | None] = mapped_column( 1ab

53 TIMESTAMP(timezone=True), nullable=True 

54 ) 

55 """Date and time when this payout was paid. Might be `None` if not yet paid.""" 1ab

56 currency: Mapped[str] = mapped_column(String(3), nullable=False) 1ab

57 """Currency of this transaction from Polar's perspective. Should be `usd`.""" 1ab

58 amount: Mapped[int] = mapped_column(BigInteger, nullable=False) 1ab

59 """Amount in cents of this transaction from Polar's perspective.""" 1ab

60 fees_amount: Mapped[int] = mapped_column(BigInteger, nullable=False) 1ab

61 """Fees amount in cents of this transaction from Polar's perspective.""" 1ab

62 account_currency: Mapped[str] = mapped_column(String(3), nullable=False) 1ab

63 """Currency of this transaction from user's account perspective. Might not be `usd`.""" 1ab

64 account_amount: Mapped[int] = mapped_column(BigInteger, nullable=False) 1ab

65 """Amount in cents of this transaction from user's account perspective.""" 1ab

66 

67 account_id: Mapped[UUID] = mapped_column( 1ab

68 Uuid, ForeignKey("accounts.id", ondelete="restrict"), nullable=False 

69 ) 

70 """ID of the `Account` concerned by this payout.""" 1ab

71 account: Mapped["Account"] = relationship("Account", lazy="raise") 1ab

72 

73 invoice_number: Mapped[str] = mapped_column(String, nullable=False) 1ab

74 """Reverse invoice number for this payout.""" 1ab

75 invoice_path: Mapped[str | None] = mapped_column( 1ab

76 String, nullable=True, default=None 

77 ) 

78 """ 1ab

79 Path to the invoice for this payout on the storage bucket. 

80 

81 Might be `None` if not yet created. 

82 """ 

83 

84 transaction: Mapped["Transaction"] = relationship( 1ab

85 "Transaction", 

86 back_populates="payout", 

87 lazy="raise", 

88 uselist=False, 

89 foreign_keys="Transaction.payout_id", 

90 ) 

91 """Transaction associated with this payout.""" 1ab

92 

93 @property 1ab

94 def gross_amount(self) -> int: 1ab

95 """Gross amount of this payout in cents.""" 

96 return self.amount + self.fees_amount 

97 

98 @property 1ab

99 def fees_transactions(self) -> list["Transaction"]: 1ab

100 """List of transactions that are fees for this payout.""" 

101 return [ 

102 transaction 

103 for transaction in self.transaction.incurred_transactions 

104 if transaction.account_id is not None 

105 ] 

106 

107 @property 1ab

108 def is_invoice_generated(self) -> bool: 1ab

109 return self.invoice_path is not None