Coverage for polar/order/tasks.py: 30%
126 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 12:42 +0000
1import uuid
3import stripe as stripe_lib
4import structlog
5from dramatiq import Retry
6from sqlalchemy.orm import joinedload
8from polar.exceptions import PolarTaskError
9from polar.logging import Logger
10from polar.models import Customer, Order
11from polar.models.order import OrderBillingReasonInternal
12from polar.payment_method.repository import PaymentMethodRepository
13from polar.product.repository import ProductRepository
14from polar.subscription.repository import SubscriptionRepository
15from polar.transaction.service.balance import PaymentTransactionForChargeDoesNotExist
16from polar.worker import (
17 AsyncSessionMaker,
18 CronTrigger,
19 TaskPriority,
20 actor,
21 can_retry,
22 enqueue_job,
23)
25from .repository import OrderRepository
26from .service import CardPaymentFailed, NoPendingBillingEntries
27from .service import order as order_service
29log: Logger = structlog.get_logger()
31MAX_RETRIES = 10
34class OrderTaskError(PolarTaskError): ... 34 ↛ 37line 34 didn't jump to line 37 because
37class SubscriptionDoesNotExist(OrderTaskError):
38 def __init__(self, subscription_id: uuid.UUID) -> None:
39 self.subscription_id = subscription_id
40 message = f"The subscription with id {subscription_id} does not exist."
41 super().__init__(message)
44class ProductDoesNotExist(OrderTaskError):
45 def __init__(self, product_id: uuid.UUID) -> None:
46 self.product_id = product_id
47 message = f"The product with id {product_id} does not exist."
48 super().__init__(message)
51class OrderDoesNotExist(OrderTaskError):
52 def __init__(self, order_id: uuid.UUID) -> None:
53 self.order_id = order_id
54 message = f"The order with id {order_id} does not exist."
55 super().__init__(message)
58class PaymentMethodDoesNotExist(OrderTaskError):
59 def __init__(self, payment_method_id: uuid.UUID) -> None:
60 self.payment_method_id = payment_method_id
61 message = f"The payment method with id {payment_method_id} does not exist."
62 super().__init__(message)
65@actor(actor_name="order.create_subscription_order", priority=TaskPriority.LOW)
66async def create_subscription_order(
67 subscription_id: uuid.UUID, order_reason: OrderBillingReasonInternal
68) -> None:
69 async with AsyncSessionMaker() as session:
70 repository = SubscriptionRepository.from_session(session)
71 subscription = await repository.get_by_id(
72 subscription_id, options=repository.get_eager_options()
73 )
74 if subscription is None:
75 raise SubscriptionDoesNotExist(subscription_id)
77 try:
78 await order_service.create_subscription_order(
79 session, subscription, order_reason
80 )
81 except NoPendingBillingEntries:
82 # Skip creating an order if there are no pending billing entries.
83 # Usually happens if the subscription is now canceled, and no usage-based billing is pending
84 pass
87@actor(actor_name="order.trigger_payment", priority=TaskPriority.LOW)
88async def trigger_payment(order_id: uuid.UUID, payment_method_id: uuid.UUID) -> None:
89 async with AsyncSessionMaker() as session:
90 repository = OrderRepository.from_session(session)
91 order = await repository.get_by_id(
92 order_id, options=repository.get_eager_options()
93 )
94 if order is None:
95 raise OrderDoesNotExist(order_id)
97 payment_method_repository = PaymentMethodRepository.from_session(session)
98 payment_method = await payment_method_repository.get_by_id_and_customer(
99 payment_method_id, order.customer_id
100 )
101 if payment_method is None:
102 log.info(
103 "Payment method not found, triggering dunning process",
104 order_id=order_id,
105 payment_method_id=payment_method_id,
106 )
107 await order_service.handle_payment_failure(session, order)
108 return
110 try:
111 await order_service.trigger_payment(session, order, payment_method)
112 except CardPaymentFailed:
113 # Card errors should not be retried - they will be handled by the dunning process
114 # Log the failure but don't retry the task
115 log.info(
116 "Card payment failed, not retrying - will be handled by dunning",
117 order_id=order_id,
118 )
119 return
120 except (
121 stripe_lib.APIConnectionError,
122 stripe_lib.APIError,
123 stripe_lib.RateLimitError,
124 ) as e:
125 # Network/availability errors should be retried
126 log.error(
127 "Stripe service error during payment trigger, retrying",
128 order_id=order_id,
129 error_type=type(e).__name__,
130 error_message=str(e),
131 )
132 if can_retry():
133 raise Retry() from e
134 else:
135 raise
138@actor(actor_name="order.balance", priority=TaskPriority.LOW)
139async def create_order_balance(order_id: uuid.UUID, charge_id: str) -> None:
140 async with AsyncSessionMaker() as session:
141 repository = OrderRepository.from_session(session)
142 order = await repository.get_by_id(
143 order_id,
144 options=(joinedload(Order.customer).joinedload(Customer.organization),),
145 )
146 if order is None:
147 raise OrderDoesNotExist(order_id)
149 try:
150 await order_service.create_order_balance(session, order, charge_id)
151 except PaymentTransactionForChargeDoesNotExist as e:
152 # Retry because Stripe webhooks order is not guaranteed,
153 # so we might not have been able to handle subscription.created
154 # or charge.succeeded yet!
155 if can_retry():
156 raise Retry() from e
157 # Raise the exception to be notified about it
158 else:
159 raise
162@actor(actor_name="order.update_product_benefits_grants", priority=TaskPriority.MEDIUM)
163async def update_product_benefits_grants(product_id: uuid.UUID) -> None:
164 async with AsyncSessionMaker() as session:
165 product_repository = ProductRepository.from_session(session)
166 product = await product_repository.get_by_id(product_id)
167 if product is None:
168 raise ProductDoesNotExist(product_id)
170 await order_service.update_product_benefits_grants(session, product)
173@actor(actor_name="order.confirmation_email", priority=TaskPriority.LOW)
174async def order_confirmation_email(order_id: uuid.UUID) -> None:
175 async with AsyncSessionMaker() as session:
176 repository = OrderRepository.from_session(session)
177 order = await repository.get_by_id(
178 order_id, options=repository.get_eager_options()
179 )
180 if order is None:
181 raise OrderDoesNotExist(order_id)
183 await order_service.send_confirmation_email(session, order)
186@actor(actor_name="order.invoice", priority=TaskPriority.LOW)
187async def order_invoice(order_id: uuid.UUID) -> None:
188 async with AsyncSessionMaker() as session:
189 repository = OrderRepository.from_session(session)
190 order = await repository.get_by_id(
191 order_id, options=repository.get_eager_options()
192 )
193 if order is None:
194 raise OrderDoesNotExist(order_id)
196 await order_service.generate_invoice(session, order)
199@actor(
200 actor_name="order.process_dunning",
201 cron_trigger=CronTrigger.from_crontab("0 * * * *"),
202 priority=TaskPriority.MEDIUM,
203)
204async def process_dunning() -> None:
205 """Process all orders that are due for dunning (payment retry)."""
206 async with AsyncSessionMaker() as session:
207 order_repository = OrderRepository.from_session(session)
208 due_orders = await order_repository.get_due_dunning_orders()
210 for order in due_orders:
211 enqueue_job("order.process_dunning_order", order.id)
214@actor(actor_name="order.process_dunning_order", priority=TaskPriority.MEDIUM)
215async def process_dunning_order(order_id: uuid.UUID) -> None:
216 """Process a single order due for dunning (payment retry)."""
217 async with AsyncSessionMaker() as session:
218 order_repository = OrderRepository.from_session(session)
219 order = await order_repository.get_by_id(
220 order_id, options=order_repository.get_eager_options()
221 )
222 if order is None:
223 raise OrderDoesNotExist(order_id)
225 await order_service.process_dunning_order(session, order)