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

1from datetime import datetime 

2from enum import StrEnum 

3from typing import TYPE_CHECKING 

4from uuid import UUID 

5 

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

7from sqlalchemy.orm import Mapped, mapped_column, relationship 

8from sqlalchemy.sql.sqltypes import BigInteger 

9 

10from polar.enums import AccountType 

11from polar.kit.db.models import RecordModel 

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

13 

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

15 from .account import Account 

16 from .transaction import Transaction 

17 

18 

19class PayoutStatus(StrEnum): 

20 pending = "pending" 

21 in_transit = "in_transit" 

22 succeeded = "succeeded" 

23 

24 @classmethod 

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

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): 

34 __tablename__ = "payouts" 

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

36 

37 processor: Mapped[AccountType] = mapped_column( 

38 StringEnum(AccountType), nullable=False 

39 ) 

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

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

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.""" 

45 status: Mapped[PayoutStatus] = mapped_column( 

46 StringEnum(PayoutStatus), 

47 nullable=False, 

48 index=True, 

49 default=PayoutStatus.pending, 

50 ) 

51 """Status of this payout.""" 

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

53 TIMESTAMP(timezone=True), nullable=True 

54 ) 

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

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

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

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

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

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

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

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

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

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

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

66 

67 account_id: Mapped[UUID] = mapped_column( 

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

69 ) 

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

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

72 

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

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

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

76 String, nullable=True, default=None 

77 ) 

78 """ 

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( 

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.""" 

92 

93 @property 

94 def gross_amount(self) -> int: 

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

96 return self.amount + self.fees_amount 

97 

98 @property 

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

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 

108 def is_invoice_generated(self) -> bool: 

109 return self.invoice_path is not None