Coverage for polar/payment/service.py: 39%

1import uuid 

2from collections.abc import Sequence 

3 

4import stripe as stripe_lib 

5 

6from polar.auth.models import AuthSubject, Organization, User 

7from polar.enums import PaymentProcessor 

8from polar.exceptions import PolarError 

9from polar.kit.pagination import PaginationParams 

10from polar.kit.sorting import Sorting 

11from polar.kit.utils import generate_uuid 

12from polar.models import Checkout, Order, Payment, Wallet 

13from polar.models.payment import PaymentStatus 

14from polar.postgres import AsyncReadSession, AsyncSession 

15 

16from .repository import PaymentRepository 

17from .sorting import PaymentSortProperty 

18 

19 

20class PaymentError(PolarError): ... 20 ↛ 23line 20 didn't jump to line 23 because

21 

22 

23class UnlinkedPaymentError(PaymentError): 

24 def __init__(self, processor_id: str) -> None: 

25 self.processor_id = processor_id 

26 message = ( 

27 f"Received a payment with id {processor_id} that is not linked " 

28 "to any checkout or order." 

29 ) 

30 super().__init__(message) 

31 

32 

33class UnhandledPaymentIntent(PaymentError): 

34 def __init__(self, payment_intent_id: str) -> None: 

35 self.payment_intent_id = payment_intent_id 

36 message = ( 

37 f"Received a payment intent with id {payment_intent_id} " 

38 "that we shouldn't handle." 

39 ) 

40 super().__init__(message) 

41 

42 

43class PaymentService: 

44 async def list( 

45 self, 

46 session: AsyncReadSession, 

47 auth_subject: AuthSubject[User | Organization], 

48 *, 

49 organization_id: Sequence[uuid.UUID] | None = None, 

50 checkout_id: Sequence[uuid.UUID] | None = None, 

51 order_id: Sequence[uuid.UUID] | None = None, 

52 status: Sequence[PaymentStatus] | None = None, 

53 method: Sequence[str] | None = None, 

54 customer_email: Sequence[str] | None = None, 

55 pagination: PaginationParams, 

56 sorting: list[Sorting[PaymentSortProperty]] = [ 

57 (PaymentSortProperty.created_at, True) 

58 ], 

59 ) -> tuple[Sequence[Payment], int]: 

60 repository = PaymentRepository.from_session(session) 

61 statement = repository.get_readable_statement(auth_subject) 

62 

63 if organization_id is not None: 63 ↛ 66line 63 didn't jump to line 66 because the condition on line 63 was always true

64 statement = statement.where(Payment.organization_id.in_(organization_id)) 

65 

66 if checkout_id is not None: 

67 statement = statement.where(Payment.checkout_id.in_(checkout_id)) 

68 

69 if order_id is not None: 

70 statement = statement.where(Payment.order_id.in_(order_id)) 

71 

72 if status is not None: 

73 statement = statement.where(Payment.status.in_(status)) 

74 

75 if method is not None: 

76 statement = statement.where(Payment.method.in_(method)) 

77 

78 if customer_email is not None: 

79 statement = statement.where(Payment.customer_email.in_(customer_email)) 

80 

81 statement = repository.apply_sorting(statement, sorting) 

82 

83 return await repository.paginate( 

84 statement, limit=pagination.limit, page=pagination.page 

85 ) 

86 

87 async def get( 

88 self, 

89 session: AsyncReadSession, 

90 auth_subject: AuthSubject[User | Organization], 

91 id: uuid.UUID, 

92 ) -> Payment | None: 

93 repository = PaymentRepository.from_session(session) 

94 statement = repository.get_readable_statement(auth_subject).where( 

95 Payment.id == id 

96 ) 

97 return await repository.get_one_or_none(statement) 

98 

99 async def upsert_from_stripe_charge( 

100 self, 

101 session: AsyncSession, 

102 charge: stripe_lib.Charge, 

103 checkout: Checkout | None, 

104 wallet: Wallet | None, 

105 order: Order | None, 

106 ) -> Payment: 

107 repository = PaymentRepository.from_session(session) 

108 

109 payment = await repository.get_by_processor_id( 

110 PaymentProcessor.stripe, charge.id 

111 ) 

112 if payment is None: 

113 payment = Payment( 

114 id=generate_uuid(), 

115 processor=PaymentProcessor.stripe, 

116 processor_id=charge.id, 

117 ) 

118 

119 payment.status = PaymentStatus.from_stripe_charge(charge.status) 

120 payment.amount = charge.amount 

121 payment.currency = charge.currency 

122 

123 payment_method_details = charge.payment_method_details 

124 assert payment_method_details is not None 

125 payment.method = payment_method_details.type 

126 payment.method_metadata = dict( 

127 payment_method_details[payment_method_details.type] 

128 ) 

129 payment.customer_email = charge.billing_details.email 

130 

131 if charge.outcome is not None: 

132 payment.decline_reason = charge.outcome.reason 

133 # Stripe also sets success message, but we don't need it 

134 if payment.decline_reason is not None: 

135 payment.decline_message = charge.outcome.seller_message 

136 

137 risk_level = charge.outcome.risk_level 

138 if risk_level is not None and risk_level != "not_assessed": 

139 payment.risk_level = risk_level 

140 payment.risk_score = charge.outcome.get("risk_score") 

141 

142 payment.checkout = checkout 

143 payment.order = order 

144 payment.wallet = wallet 

145 

146 if checkout is not None: 

147 payment.organization = checkout.organization 

148 elif wallet is not None: 

149 payment.organization = wallet.organization 

150 elif order is not None: 

151 payment.organization = order.organization 

152 else: 

153 raise UnlinkedPaymentError(charge.id) 

154 

155 return await repository.update(payment) 

156 

157 async def upsert_from_stripe_payment_intent( 

158 self, 

159 session: AsyncSession, 

160 payment_intent: stripe_lib.PaymentIntent, 

161 checkout: Checkout | None, 

162 order: Order | None, 

163 ) -> Payment: 

164 # Only handle payment intents that are not linked to a charge, and which 

165 # have a last_payment_error. 

166 # It's the case when the payment method fails authentication, like 3DS. 

167 # In other cases, we handle it through the charge (see above). 

168 if ( 

169 payment_intent.latest_charge is not None 

170 or payment_intent.last_payment_error is None 

171 ): 

172 raise UnhandledPaymentIntent(payment_intent.id) 

173 

174 repository = PaymentRepository.from_session(session) 

175 

176 payment = await repository.get_by_processor_id( 

177 PaymentProcessor.stripe, payment_intent.id 

178 ) 

179 if payment is None: 

180 payment = Payment( 

181 id=generate_uuid(), 

182 processor=PaymentProcessor.stripe, 

183 processor_id=payment_intent.id, 

184 ) 

185 

186 payment.status = PaymentStatus.failed 

187 payment.amount = payment_intent.amount 

188 payment.currency = payment_intent.currency 

189 

190 payment_error = payment_intent.last_payment_error 

191 payment_method = payment_error.payment_method 

192 assert payment_method is not None 

193 payment.method = payment_method.type 

194 payment.method_metadata = dict(payment_method[payment_method.type]) 

195 payment.customer_email = payment_intent.receipt_email 

196 

197 payment.decline_reason = getattr(payment_error, "code", None) 

198 payment.decline_message = payment_error.message 

199 

200 payment.checkout = checkout 

201 payment.order = order 

202 

203 if checkout is not None: 

204 payment.organization = checkout.organization 

205 elif order is not None: 

206 payment.organization = order.organization 

207 else: 

208 raise UnlinkedPaymentError(payment_intent.id) 

209 

210 return await repository.update(payment) 

211 

212 

213payment = PaymentService()