Coverage for polar/payment/service.py: 19%
106 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 16:17 +0000
1import uuid 1a
2from collections.abc import Sequence 1a
4import stripe as stripe_lib 1a
6from polar.auth.models import AuthSubject, Organization, User 1a
7from polar.enums import PaymentProcessor 1a
8from polar.exceptions import PolarError 1a
9from polar.kit.pagination import PaginationParams 1a
10from polar.kit.sorting import Sorting 1a
11from polar.kit.utils import generate_uuid 1a
12from polar.models import Checkout, Order, Payment, Wallet 1a
13from polar.models.payment import PaymentStatus 1a
14from polar.postgres import AsyncReadSession, AsyncSession 1a
16from .repository import PaymentRepository 1a
17from .sorting import PaymentSortProperty 1a
20class PaymentError(PolarError): ... 1a
23class UnlinkedPaymentError(PaymentError): 1a
24 def __init__(self, processor_id: str) -> None: 1a
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)
33class UnhandledPaymentIntent(PaymentError): 1a
34 def __init__(self, payment_intent_id: str) -> None: 1a
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)
43class PaymentService: 1a
44 async def list( 1a
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)
63 if organization_id is not None:
64 statement = statement.where(Payment.organization_id.in_(organization_id))
66 if checkout_id is not None:
67 statement = statement.where(Payment.checkout_id.in_(checkout_id))
69 if order_id is not None:
70 statement = statement.where(Payment.order_id.in_(order_id))
72 if status is not None:
73 statement = statement.where(Payment.status.in_(status))
75 if method is not None:
76 statement = statement.where(Payment.method.in_(method))
78 if customer_email is not None:
79 statement = statement.where(Payment.customer_email.in_(customer_email))
81 statement = repository.apply_sorting(statement, sorting)
83 return await repository.paginate(
84 statement, limit=pagination.limit, page=pagination.page
85 )
87 async def get( 1a
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)
99 async def upsert_from_stripe_charge( 1a
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)
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 )
119 payment.status = PaymentStatus.from_stripe_charge(charge.status)
120 payment.amount = charge.amount
121 payment.currency = charge.currency
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
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
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")
142 payment.checkout = checkout
143 payment.order = order
144 payment.wallet = wallet
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)
155 return await repository.update(payment)
157 async def upsert_from_stripe_payment_intent( 1a
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)
174 repository = PaymentRepository.from_session(session)
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 )
186 payment.status = PaymentStatus.failed
187 payment.amount = payment_intent.amount
188 payment.currency = payment_intent.currency
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
197 payment.decline_reason = getattr(payment_error, "code", None)
198 payment.decline_message = payment_error.message
200 payment.checkout = checkout
201 payment.order = order
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)
210 return await repository.update(payment)
213payment = PaymentService() 1a