Coverage for polar/models/refund.py: 63%

1from enum import StrEnum 

2from typing import TYPE_CHECKING, Any, Literal 

3from uuid import UUID 

4 

5from sqlalchemy import ( 

6 Boolean, 

7 ColumnElement, 

8 ForeignKey, 

9 Integer, 

10 String, 

11 Uuid, 

12 type_coerce, 

13) 

14from sqlalchemy.dialects.postgresql import JSONB 

15from sqlalchemy.ext.hybrid import hybrid_property 

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

17 

18from polar.enums import PaymentProcessor 

19from polar.kit.db.models import RecordModel 

20from polar.kit.metadata import MetadataMixin 

21 

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

23 from polar.models import ( 

24 Customer, 

25 Order, 

26 Organization, 

27 Pledge, 

28 Subscription, 

29 ) 

30 

31 

32class RefundStatus(StrEnum): 

33 pending = "pending" 

34 succeeded = "succeeded" 

35 failed = "failed" 

36 canceled = "canceled" 

37 

38 

39# Decoupled from Stripe 

40# 1) Allowing more reasons (good signals) 

41# 2) Allowing us to enable merchants to set `fraudulent` without automatically 

42# tagging it as such on Stripe. 

43class RefundReason(StrEnum): 

44 duplicate = "duplicate" 

45 fraudulent = "fraudulent" 

46 customer_request = "customer_request" 

47 service_disruption = "service_disruption" 

48 satisfaction_guarantee = "satisfaction_guarantee" 

49 other = "other" 

50 

51 @classmethod 

52 def from_stripe( 

53 cls, 

54 reason: ( 

55 Literal[ 

56 "duplicate", 

57 "expired_uncaptured_charge", 

58 "fraudulent", 

59 "requested_by_customer", 

60 ] 

61 | None 

62 ), 

63 ) -> "RefundReason": 

64 if reason == "requested_by_customer": 

65 return cls.customer_request 

66 elif reason == "fraudulent": 

67 return cls.fraudulent 

68 elif reason == "duplicate": 

69 return cls.duplicate 

70 return cls.other 

71 

72 @classmethod 

73 def to_stripe( 

74 cls, reason: "RefundReason" 

75 ) -> Literal["requested_by_customer", "duplicate"]: 

76 if reason == cls.duplicate: 

77 return "duplicate" 

78 

79 # Avoid directly setting fraudulent since that blocks customers and can 

80 # be abused, i.e we should monitor our own fraudulent status and set it 

81 # retroactively on Stripe. 

82 return "requested_by_customer" 

83 

84 

85class RefundFailureReason(StrEnum): 

86 unknown = "unknown" 

87 declined = "declined" 

88 card_expired = "card_expired" 

89 card_lost = "card_lost" 

90 disputed = "disputed" 

91 insufficient_funds = "insufficient_funds" 

92 merchant_request = "merchant_request" 

93 

94 @classmethod 

95 def from_stripe( 

96 cls, 

97 reason: ( 

98 Literal[ 

99 "lost_or_stolen_card", 

100 "expired_or_canceled_card", 

101 "charge_for_pending_refund_disputed", 

102 "insufficient_funds", 

103 "merchant_request", 

104 "unknown", 

105 ] 

106 | None 

107 ), 

108 ) -> "RefundFailureReason | None": 

109 if reason is None: 

110 return None 

111 

112 if reason == "lost_or_stolen_card": 

113 return cls.card_lost 

114 elif reason == "expired_or_canceled_card": 

115 return cls.card_expired 

116 elif reason == "charge_for_pending_refund_disputed": 

117 return cls.disputed 

118 elif reason == "insufficient_funds": 

119 return cls.insufficient_funds 

120 elif reason == "merchant_request": 

121 return cls.merchant_request 

122 return cls.unknown 

123 

124 

125class Refund(MetadataMixin, RecordModel): 

126 __tablename__ = "refunds" 

127 

128 status: Mapped[RefundStatus] = mapped_column(String, nullable=False) 

129 reason: Mapped[RefundReason] = mapped_column(String, nullable=False) 

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

131 tax_amount: Mapped[int] = mapped_column(Integer, nullable=False) 

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

133 

134 comment: Mapped[str | None] = mapped_column(String, nullable=True) 

135 

136 failure_reason: Mapped[RefundFailureReason | None] = mapped_column( 

137 String, nullable=True 

138 ) 

139 

140 destination_details: Mapped[dict[str, Any]] = mapped_column( 

141 JSONB, nullable=False, default=dict 

142 ) 

143 

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

145 Uuid, ForeignKey("orders.id"), nullable=True, index=True 

146 ) 

147 

148 @declared_attr 

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

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

151 

152 subscription_id: Mapped[UUID | None] = mapped_column( 

153 Uuid, ForeignKey("subscriptions.id"), nullable=True, index=True 

154 ) 

155 

156 @declared_attr 

157 def subscription(cls) -> Mapped["Subscription | None"]: 

158 return relationship("Subscription", lazy="raise") 

159 

160 organization_id: Mapped[UUID | None] = mapped_column( 

161 Uuid, ForeignKey("organizations.id"), nullable=True, index=True 

162 ) 

163 

164 @declared_attr 

165 def organization(cls) -> Mapped["Organization | None"]: 

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

167 

168 customer_id: Mapped[UUID | None] = mapped_column( 

169 Uuid, ForeignKey("customers.id"), nullable=True, index=True 

170 ) 

171 

172 @declared_attr 

173 def customer(cls) -> Mapped["Customer | None"]: 

174 return relationship("Customer", lazy="raise") 

175 

176 pledge_id: Mapped[UUID | None] = mapped_column( 

177 Uuid, 

178 ForeignKey("pledges.id"), 

179 nullable=True, 

180 ) 

181 

182 @declared_attr 

183 def pledge(cls) -> Mapped["Pledge | None"]: 

184 return relationship("Pledge", lazy="raise") 

185 

186 # Created refund was set to revoke customer benefits? 

187 revoke_benefits: Mapped[bool] = mapped_column( 

188 Boolean, 

189 nullable=False, 

190 default=False, 

191 ) 

192 

193 processor: Mapped[PaymentProcessor] = mapped_column( 

194 String, 

195 nullable=False, 

196 ) 

197 processor_id: Mapped[str] = mapped_column( 

198 String, 

199 nullable=False, 

200 unique=True, 

201 index=True, 

202 ) 

203 processor_reason: Mapped[str] = mapped_column(String, nullable=False) 

204 processor_receipt_number: Mapped[str | None] = mapped_column(String, nullable=True) 

205 processor_balance_transaction_id: Mapped[str | None] = mapped_column( 

206 String, nullable=True 

207 ) 

208 

209 tax_transaction_processor_id: Mapped[str | None] = mapped_column( 

210 String, nullable=True, default=None 

211 ) 

212 

213 @hybrid_property 

214 def succeeded(self) -> bool: 

215 return self.status == RefundStatus.succeeded 

216 

217 @succeeded.inplace.expression 

218 @classmethod 

219 def _succeeded_expression(cls) -> ColumnElement[bool]: 

220 return type_coerce( 

221 cls.status.in_(RefundStatus.succeeded), 

222 Boolean, 

223 ) 

224 

225 @hybrid_property 

226 def total_amount(self) -> int: 

227 return self.amount + self.tax_amount 

228 

229 @total_amount.inplace.expression 

230 @classmethod 

231 def _total_amount_expression(cls) -> ColumnElement[int]: 

232 return cls.amount + cls.tax_amount