Coverage for polar/transaction/tasks.py: 44%
30 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 17:15 +0000
1import uuid 1a
3from dramatiq import Retry 1a
5from polar.exceptions import PolarTaskError 1a
6from polar.worker import AsyncSessionMaker, CronTrigger, TaskPriority, actor, can_retry 1a
8from .repository import PaymentTransactionRepository 1a
9from .service.processor_fee import ( 1a
10 BalanceTransactionNotFound,
11)
12from .service.processor_fee import ( 1a
13 processor_fee_transaction as processor_fee_transaction_service,
14)
17class TransactionTaskError(PolarTaskError): ... 1a
20class PaymentTransactionDoesNotExist(TransactionTaskError): 1a
21 def __init__(self, payment_transaction_id: uuid.UUID) -> None: 1a
22 self.payment_transaction_id = payment_transaction_id
23 message = (
24 f"Payment transaction with id {payment_transaction_id} does not exist."
25 )
26 super().__init__(message)
29@actor( 1a
30 actor_name="processor_fee.sync_stripe_fees",
31 cron_trigger=CronTrigger(hour=0, minute=0),
32 priority=TaskPriority.LOW,
33)
34async def sync_stripe_fees() -> None: 1a
35 async with AsyncSessionMaker() as session:
36 await processor_fee_transaction_service.sync_stripe_fees(session)
39@actor(actor_name="processor_fee.create_payment_fees", priority=TaskPriority.LOW) 1a
40async def create_payment_fees(payment_transaction_id: uuid.UUID) -> None: 1a
41 async with AsyncSessionMaker() as session:
42 repository = PaymentTransactionRepository.from_session(session)
43 payment_transaction = await repository.get_by_id(payment_transaction_id)
44 if payment_transaction is None:
45 raise PaymentTransactionDoesNotExist(payment_transaction_id)
47 try:
48 await processor_fee_transaction_service.create_payment_fees(
49 session, payment_transaction=payment_transaction
50 )
51 except BalanceTransactionNotFound as e:
52 # Retry because Stripe may have not created the balance transaction yet
53 if can_retry():
54 raise Retry() from e
55 # Raise the exception to be notified about it
56 else:
57 raise