Coverage for polar/payment_method/service.py: 22%

1import uuid 

2 

3import stripe as stripe_lib 

4from sqlalchemy import select 

5 

6from polar.customer.repository import CustomerRepository 

7from polar.enums import PaymentProcessor 

8from polar.exceptions import PolarError 

9from polar.integrations.stripe.service import stripe as stripe_service 

10from polar.integrations.stripe.utils import get_expandable_id 

11from polar.models import Checkout, Customer, Order, PaymentMethod, Subscription 

12from polar.models.subscription import SubscriptionStatus 

13from polar.postgres import AsyncSession 

14 

15from .repository import PaymentMethodRepository 

16 

17 

18class PaymentMethodError(PolarError): ... 18 ↛ 21line 18 didn't jump to line 21 because

19 

20 

21class NoPaymentMethodOnIntent(PaymentMethodError): 

22 def __init__(self, intent_id: str) -> None: 

23 self.intent_id = intent_id 

24 message = f"No payment method found on Stripe intent with ID {intent_id}." 

25 super().__init__(message) 

26 

27 

28class PaymentMethodInUseByActiveSubscription(PaymentMethodError): 

29 def __init__(self, subscription_ids: list[uuid.UUID]) -> None: 

30 self.subscription_ids = subscription_ids 

31 message = ( 

32 "Cannot delete payment method. It is currently used by active " 

33 "subscription and no alternative payment methods " 

34 ) 

35 super().__init__(message, 400) 

36 

37 

38class PaymentMethodService: 

39 async def upsert_from_stripe( 

40 self, 

41 session: AsyncSession, 

42 customer: Customer, 

43 stripe_payment_method: stripe_lib.PaymentMethod, 

44 *, 

45 flush: bool = False, 

46 ) -> PaymentMethod: 

47 repository = PaymentMethodRepository.from_session(session) 

48 

49 payment_method = await repository.get_by_customer_and_processor_id( 

50 customer.id, 

51 PaymentProcessor.stripe, 

52 stripe_payment_method.id, 

53 include_deleted=True, 

54 options=repository.get_eager_options(), 

55 ) 

56 if payment_method is None: 

57 payment_method = PaymentMethod( 

58 processor=PaymentProcessor.stripe, 

59 processor_id=stripe_payment_method.id, 

60 customer=customer, 

61 ) 

62 

63 payment_method.type = stripe_payment_method.type 

64 payment_method.method_metadata = stripe_payment_method[ 

65 stripe_payment_method.type 

66 ] 

67 payment_method.deleted_at = None # Restore if it was soft-deleted 

68 

69 return await repository.update(payment_method, flush=flush) 

70 

71 async def upsert_from_stripe_intent( 

72 self, 

73 session: AsyncSession, 

74 intent: stripe_lib.Charge | stripe_lib.SetupIntent, 

75 checkout: Checkout, 

76 ) -> PaymentMethod: 

77 if intent.payment_method is None: 

78 raise NoPaymentMethodOnIntent(intent.id) 

79 

80 stripe_payment_method = await stripe_service.get_payment_method( 

81 get_expandable_id(intent.payment_method) 

82 ) 

83 

84 assert checkout.customer is not None 

85 return await self.upsert_from_stripe( 

86 session, checkout.customer, stripe_payment_method 

87 ) 

88 

89 async def upsert_from_stripe_payment_intent_for_order( 

90 self, 

91 session: AsyncSession, 

92 payment_intent: stripe_lib.PaymentIntent, 

93 order: Order, 

94 ) -> PaymentMethod | None: 

95 """ 

96 Upsert payment method from PaymentIntent for order retry payments. 

97 Only saves if the order is for a recurring product and has a payment method attached. 

98 """ 

99 if payment_intent.payment_method is None: 

100 return None 

101 

102 if order.product and not order.product.is_recurring: 

103 return None 

104 

105 stripe_payment_method = await stripe_service.get_payment_method( 

106 get_expandable_id(payment_intent.payment_method) 

107 ) 

108 

109 customer_repository = CustomerRepository.from_session(session) 

110 customer = await customer_repository.get_by_id( 

111 order.customer_id, include_deleted=True 

112 ) 

113 assert customer is not None 

114 

115 return await self.upsert_from_stripe(session, customer, stripe_payment_method) 

116 

117 async def get_customer_payment_method( 

118 self, session: AsyncSession, customer: Customer 

119 ) -> PaymentMethod | None: 

120 repository = PaymentMethodRepository.from_session(session) 

121 if customer.default_payment_method_id is not None: 

122 return await repository.get_by_id( 

123 customer.default_payment_method_id, 

124 options=repository.get_eager_options(), 

125 ) 

126 

127 payment_methods = await repository.list_by_customer( 

128 customer.id, options=repository.get_eager_options() 

129 ) 

130 if len(payment_methods) > 0: 

131 return payment_methods[0] 

132 

133 return None 

134 

135 async def _get_active_subscription_ids( 

136 self, 

137 session: AsyncSession, 

138 payment_method: PaymentMethod, 

139 ) -> list[uuid.UUID]: 

140 repository = PaymentMethodRepository.from_session(session) 

141 stmt = select(Subscription.id).where( 

142 Subscription.payment_method_id == payment_method.id, 

143 Subscription.status.in_(SubscriptionStatus.active_statuses()), 

144 ) 

145 result = await session.execute(stmt) 

146 return [row[0] for row in result.fetchall()] 

147 

148 async def _get_alternative_payment_method( 

149 self, 

150 session: AsyncSession, 

151 payment_method: PaymentMethod, 

152 ) -> PaymentMethod | None: 

153 repository = PaymentMethodRepository.from_session(session) 

154 alternative_methods = await repository.list_by_customer( 

155 payment_method.customer_id, 

156 exclude_id=payment_method.id, 

157 ) 

158 

159 if not alternative_methods: 

160 return None 

161 

162 # Prefer the customer's default payment method if it's different from the one being deleted 

163 stmt = select(Customer.default_payment_method_id).where( 

164 Customer.id == payment_method.customer_id 

165 ) 

166 result = await session.execute(stmt) 

167 default_pm_id = result.scalar_one_or_none() 

168 

169 if default_pm_id and default_pm_id != payment_method.id: 

170 for method in alternative_methods: 

171 if method.id == default_pm_id: 

172 return method 

173 

174 # Otherwise, return the first available alternative 

175 return alternative_methods[0] 

176 

177 async def _reassign_subscriptions_payment_method( 

178 self, 

179 session: AsyncSession, 

180 from_payment_method: PaymentMethod, 

181 to_payment_method: PaymentMethod, 

182 subscription_ids: list[uuid.UUID], 

183 ) -> None: 

184 stmt = select(Subscription).where(Subscription.id.in_(subscription_ids)) 

185 result = await session.execute(stmt) 

186 subscriptions = list(result.scalars().all()) 

187 

188 for subscription in subscriptions: 

189 subscription.payment_method = to_payment_method 

190 

191 if ( 

192 subscription.stripe_subscription_id 

193 and to_payment_method.processor == PaymentProcessor.stripe 

194 ): 

195 await stripe_service.set_automatically_charged_subscription( 

196 subscription.stripe_subscription_id, 

197 payment_method=to_payment_method.processor_id, 

198 ) 

199 

200 await session.flush() 

201 

202 async def delete( 

203 self, 

204 session: AsyncSession, 

205 payment_method: PaymentMethod, 

206 ) -> None: 

207 active_subscription_ids = await self._get_active_subscription_ids( 

208 session, payment_method 

209 ) 

210 

211 if active_subscription_ids: 

212 alternative_payment_method = await self._get_alternative_payment_method( 

213 session, payment_method 

214 ) 

215 

216 if alternative_payment_method: 

217 await self._reassign_subscriptions_payment_method( 

218 session, 

219 from_payment_method=payment_method, 

220 to_payment_method=alternative_payment_method, 

221 subscription_ids=active_subscription_ids, 

222 ) 

223 else: 

224 # No alternative payment method available, raise exception 

225 raise PaymentMethodInUseByActiveSubscription(active_subscription_ids) 

226 

227 if payment_method.processor == PaymentProcessor.stripe: 

228 await stripe_service.delete_payment_method(payment_method.processor_id) 

229 

230 repository = PaymentMethodRepository.from_session(session) 

231 await repository.soft_delete(payment_method) 

232 

233 

234payment_method = PaymentMethodService()