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

1from enum import StrEnum 2 ctx1ab

2from typing import TYPE_CHECKING, Any, Literal 2 ctx1ab

3from uuid import UUID 2 ctx1ab

4 

5from sqlalchemy import ( 2 ctx1ab

6 Boolean, 

7 ColumnElement, 

8 ForeignKey, 

9 Integer, 

10 String, 

11 Uuid, 

12 type_coerce, 

13) 

14from sqlalchemy.dialects.postgresql import JSONB 2 ctx1ab

15from sqlalchemy.ext.hybrid import hybrid_property 2 ctx1ab

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

17 

18from polar.enums import PaymentProcessor 2 ctx1ab

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

20from polar.kit.metadata import MetadataMixin 2 ctx1ab

21 

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

23 from polar.models import ( 

24 Customer, 

25 Order, 

26 Organization, 

27 Pledge, 

28 Subscription, 

29 ) 

30 

31 

32class RefundStatus(StrEnum): 2 ctx1ab

33 pending = "pending" 2 ctx1ab

34 succeeded = "succeeded" 2 ctx1ab

35 failed = "failed" 2 ctx1ab

36 canceled = "canceled" 2 ctx1ab

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

44 duplicate = "duplicate" 2 ctx1ab

45 fraudulent = "fraudulent" 2 ctx1ab

46 customer_request = "customer_request" 2 ctx1ab

47 service_disruption = "service_disruption" 2 ctx1ab

48 satisfaction_guarantee = "satisfaction_guarantee" 2 ctx1ab

49 other = "other" 2 ctx1ab

50 

51 @classmethod 2 ctx1ab

52 def from_stripe( 2 ctx1ab

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

73 def to_stripe( 2 ctx1ab

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

86 unknown = "unknown" 2 ctx1ab

87 declined = "declined" 2 ctx1ab

88 card_expired = "card_expired" 2 ctx1ab

89 card_lost = "card_lost" 2 ctx1ab

90 disputed = "disputed" 2 ctx1ab

91 insufficient_funds = "insufficient_funds" 2 ctx1ab

92 merchant_request = "merchant_request" 2 ctx1ab

93 

94 @classmethod 2 ctx1ab

95 def from_stripe( 2 ctx1ab

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

126 __tablename__ = "refunds" 2 ctx1ab

127 

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

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

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

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

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

133 

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

135 

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

137 String, nullable=True 

138 ) 

139 

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

141 JSONB, nullable=False, default=dict 

142 ) 

143 

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

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

146 ) 

147 

148 @declared_attr 2 ctx1ab

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

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

151 

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

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

154 ) 

155 

156 @declared_attr 2 ctx1ab

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

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

159 

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

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

162 ) 

163 

164 @declared_attr 2 ctx1ab

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

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

167 

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

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

170 ) 

171 

172 @declared_attr 2 ctx1ab

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

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

175 

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

177 Uuid, 

178 ForeignKey("pledges.id"), 

179 nullable=True, 

180 ) 

181 

182 @declared_attr 2 ctx1ab

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

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

185 

186 # Created refund was set to revoke customer benefits? 

187 revoke_benefits: Mapped[bool] = mapped_column( 2 ctx1ab

188 Boolean, 

189 nullable=False, 

190 default=False, 

191 ) 

192 

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

194 String, 

195 nullable=False, 

196 ) 

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

198 String, 

199 nullable=False, 

200 unique=True, 

201 index=True, 

202 ) 

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

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

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

206 String, nullable=True 

207 ) 

208 

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

210 String, nullable=True, default=None 

211 ) 

212 

213 @hybrid_property 2 ctx1ab

214 def succeeded(self) -> bool: 2 ctx1ab

215 return self.status == RefundStatus.succeeded 

216 

217 @succeeded.inplace.expression 2 ctx1ab

218 @classmethod 2 ctx1ab

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

220 return type_coerce( 

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

222 Boolean, 

223 ) 

224 

225 @hybrid_property 2 ctx1ab

226 def total_amount(self) -> int: 2 ctx1ab

227 return self.amount + self.tax_amount 

228 

229 @total_amount.inplace.expression 2 ctx1ab

230 @classmethod 2 ctx1ab

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

232 return cls.amount + cls.tax_amount