Coverage for polar/payment_method/service.py: 23%
96 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 16:17 +0000
1import uuid 1a
3import stripe as stripe_lib 1a
4from sqlalchemy import select 1a
6from polar.customer.repository import CustomerRepository 1a
7from polar.enums import PaymentProcessor 1a
8from polar.exceptions import PolarError 1a
9from polar.integrations.stripe.service import stripe as stripe_service 1a
10from polar.integrations.stripe.utils import get_expandable_id 1a
11from polar.models import Checkout, Customer, Order, PaymentMethod, Subscription 1a
12from polar.models.subscription import SubscriptionStatus 1a
13from polar.postgres import AsyncSession 1a
15from .repository import PaymentMethodRepository 1a
18class PaymentMethodError(PolarError): ... 1a
21class NoPaymentMethodOnIntent(PaymentMethodError): 1a
22 def __init__(self, intent_id: str) -> None: 1a
23 self.intent_id = intent_id
24 message = f"No payment method found on Stripe intent with ID {intent_id}."
25 super().__init__(message)
28class PaymentMethodInUseByActiveSubscription(PaymentMethodError): 1a
29 def __init__(self, subscription_ids: list[uuid.UUID]) -> None: 1a
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)
38class PaymentMethodService: 1a
39 async def upsert_from_stripe( 1a
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)
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 )
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
69 return await repository.update(payment_method, flush=flush)
71 async def upsert_from_stripe_intent( 1a
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)
80 stripe_payment_method = await stripe_service.get_payment_method(
81 get_expandable_id(intent.payment_method)
82 )
84 assert checkout.customer is not None
85 return await self.upsert_from_stripe(
86 session, checkout.customer, stripe_payment_method
87 )
89 async def upsert_from_stripe_payment_intent_for_order( 1a
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
102 if order.product and not order.product.is_recurring:
103 return None
105 stripe_payment_method = await stripe_service.get_payment_method(
106 get_expandable_id(payment_intent.payment_method)
107 )
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
115 return await self.upsert_from_stripe(session, customer, stripe_payment_method)
117 async def get_customer_payment_method( 1a
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 )
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]
133 return None
135 async def _get_active_subscription_ids( 1a
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()]
148 async def _get_alternative_payment_method( 1a
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 )
159 if not alternative_methods:
160 return None
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()
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
174 # Otherwise, return the first available alternative
175 return alternative_methods[0]
177 async def _reassign_subscriptions_payment_method( 1a
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())
188 for subscription in subscriptions:
189 subscription.payment_method = to_payment_method
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 )
200 await session.flush()
202 async def delete( 1a
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 )
211 if active_subscription_ids:
212 alternative_payment_method = await self._get_alternative_payment_method(
213 session, payment_method
214 )
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)
227 if payment_method.processor == PaymentProcessor.stripe:
228 await stripe_service.delete_payment_method(payment_method.processor_id)
230 repository = PaymentMethodRepository.from_session(session)
231 await repository.soft_delete(payment_method)
234payment_method = PaymentMethodService() 1a