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

1from enum import StrEnum 2 ctx1ab

2from typing import TYPE_CHECKING, Any, Literal, Self 2 ctx1ab

3from uuid import UUID 2 ctx1ab

4 

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

6from sqlalchemy.dialects.postgresql import JSONB 2 ctx1ab

7from sqlalchemy.ext.hybrid import hybrid_property 2 ctx1ab

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

9from sqlalchemy.sql.sqltypes import Integer 2 ctx1ab

10 

11from polar.enums import PaymentProcessor 2 ctx1ab

12from polar.kit.db.models import RecordModel 2 ctx1ab

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

14 

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

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

23 pending = "pending" 2 ctx1ab

24 succeeded = "succeeded" 2 ctx1ab

25 failed = "failed" 2 ctx1ab

26 

27 @classmethod 2 ctx1ab

28 def from_stripe_charge( 2 ctx1ab

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

30 ) -> Self: 

31 return cls(stripe_status) 

32 

33 

34class Payment(RecordModel): 2 ctx1ab

35 __tablename__ = "payments" 2 ctx1ab

36 

37 processor: Mapped[PaymentProcessor] = mapped_column( 2 ctx1ab

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

39 ) 

40 status: Mapped[PaymentStatus] = mapped_column( 2 ctx1ab

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

42 ) 

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

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

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

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

47 JSONB, nullable=False, default=dict 

48 ) 

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

50 JSONB, nullable=False, default=dict 

51 ) 

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

53 String, nullable=True, index=True 

54 ) 

55 

56 processor_id: Mapped[str] = mapped_column( 2 ctx1ab

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

58 ) 

59 

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

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

62 

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

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

65 

66 organization_id: Mapped[UUID] = mapped_column( 2 ctx1ab

67 Uuid, 

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

69 nullable=False, 

70 index=True, 

71 ) 

72 

73 @declared_attr 2 ctx1ab

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

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

76 

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

78 Uuid, 

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

80 nullable=True, 

81 index=True, 

82 ) 

83 

84 @declared_attr 2 ctx1ab

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

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

87 

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

89 Uuid, 

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

91 nullable=True, 

92 index=True, 

93 ) 

94 

95 @declared_attr 2 ctx1ab

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

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

98 

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

100 Uuid, 

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

102 nullable=True, 

103 index=True, 

104 ) 

105 

106 @declared_attr 2 ctx1ab

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

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

109 

110 @hybrid_property 2 ctx1ab

111 def is_succeeded(self) -> bool: 2 ctx1ab

112 return self.status == PaymentStatus.succeeded 

113 

114 @is_succeeded.inplace.expression 2 ctx1ab

115 @classmethod 2 ctx1ab

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

117 return cls.status == PaymentStatus.succeeded 

118 

119 @hybrid_property 2 ctx1ab

120 def is_failed(self) -> bool: 2 ctx1ab

121 return self.status == PaymentStatus.failed 

122 

123 @is_failed.inplace.expression 2 ctx1ab

124 @classmethod 2 ctx1ab

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

126 return cls.status == PaymentStatus.failed