Coverage for polar/models/payment.py: 86%

1from enum import StrEnum 

2from typing import TYPE_CHECKING, Any, Literal, Self 

3from uuid import UUID 

4 

5from sqlalchemy import ColumnElement, ForeignKey, SmallInteger, String, Uuid 

6from sqlalchemy.dialects.postgresql import JSONB 

7from sqlalchemy.ext.hybrid import hybrid_property 

8from sqlalchemy.orm import Mapped, declared_attr, mapped_column, relationship 

9from sqlalchemy.sql.sqltypes import Integer 

10 

11from polar.enums import PaymentProcessor 

12from polar.kit.db.models import RecordModel 

13from polar.kit.extensions.sqlalchemy.types import StrEnumType 

14 

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

16 from .checkout import Checkout 

17 from .order import Order 

18 from .organization import Organization 

19 from .wallet import Wallet 

20 

21 

22class PaymentStatus(StrEnum): 

23 pending = "pending" 

24 succeeded = "succeeded" 

25 failed = "failed" 

26 

27 @classmethod 

28 def from_stripe_charge( 

29 cls, stripe_status: Literal["failed", "pending", "succeeded"] 

30 ) -> Self: 

31 return cls(stripe_status) 

32 

33 

34class Payment(RecordModel): 

35 __tablename__ = "payments" 

36 

37 processor: Mapped[PaymentProcessor] = mapped_column( 

38 StrEnumType(PaymentProcessor), index=True, nullable=False 

39 ) 

40 status: Mapped[PaymentStatus] = mapped_column( 

41 StrEnumType(PaymentStatus), index=True, nullable=False 

42 ) 

43 amount: Mapped[int] = mapped_column(Integer, nullable=False) 

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

45 method: Mapped[str] = mapped_column(String, index=True, nullable=False) 

46 method_metadata: Mapped[dict[str, Any]] = mapped_column( 

47 JSONB, nullable=False, default=dict 

48 ) 

49 processor_metadata: Mapped[dict[str, Any]] = mapped_column( 

50 JSONB, nullable=False, default=dict 

51 ) 

52 customer_email: Mapped[str | None] = mapped_column( 

53 String, nullable=True, index=True 

54 ) 

55 

56 processor_id: Mapped[str] = mapped_column( 

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

58 ) 

59 

60 decline_reason: Mapped[str | None] = mapped_column(String, nullable=True) 

61 decline_message: Mapped[str | None] = mapped_column(String, nullable=True) 

62 

63 risk_level: Mapped[str | None] = mapped_column(String, nullable=True) 

64 risk_score: Mapped[int | None] = mapped_column(SmallInteger, nullable=True) 

65 

66 organization_id: Mapped[UUID] = mapped_column( 

67 Uuid, 

68 ForeignKey("organizations.id", ondelete="cascade"), 

69 nullable=False, 

70 index=True, 

71 ) 

72 

73 @declared_attr 

74 def organization(cls) -> Mapped["Organization"]: 

75 return relationship("Organization", lazy="raise") 

76 

77 checkout_id: Mapped[UUID | None] = mapped_column( 

78 Uuid, 

79 ForeignKey("checkouts.id", ondelete="set null"), 

80 nullable=True, 

81 index=True, 

82 ) 

83 

84 @declared_attr 

85 def checkout(cls) -> Mapped["Checkout | None"]: 

86 return relationship("Checkout", lazy="raise") 

87 

88 wallet_id: Mapped[UUID | None] = mapped_column( 

89 Uuid, 

90 ForeignKey("wallets.id", ondelete="set null"), 

91 nullable=True, 

92 index=True, 

93 ) 

94 

95 @declared_attr 

96 def wallet(cls) -> Mapped["Wallet | None"]: 

97 return relationship("Wallet", lazy="raise") 

98 

99 order_id: Mapped[UUID | None] = mapped_column( 

100 Uuid, 

101 ForeignKey("orders.id", ondelete="set null"), 

102 nullable=True, 

103 index=True, 

104 ) 

105 

106 @declared_attr 

107 def order(cls) -> Mapped["Order | None"]: 

108 return relationship("Order", lazy="raise") 

109 

110 @hybrid_property 

111 def is_succeeded(self) -> bool: 

112 return self.status == PaymentStatus.succeeded 

113 

114 @is_succeeded.inplace.expression 

115 @classmethod 

116 def _is_succeeded_expression(cls) -> ColumnElement[bool]: 

117 return cls.status == PaymentStatus.succeeded 

118 

119 @hybrid_property 

120 def is_failed(self) -> bool: 

121 return self.status == PaymentStatus.failed 

122 

123 @is_failed.inplace.expression 

124 @classmethod 

125 def _is_failed_expression(cls) -> ColumnElement[bool]: 

126 return cls.status == PaymentStatus.failed