Coverage for polar/transaction/service/processor_fee.py: 13%
114 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 16:17 +0000
1from datetime import UTC, datetime 1a
2from typing import Literal 1a
4from polar.enums import PaymentProcessor 1a
5from polar.integrations.stripe.service import stripe as stripe_service 1a
6from polar.integrations.stripe.utils import get_expandable_id 1a
7from polar.models import Refund, Transaction 1a
8from polar.models.transaction import Processor, ProcessorFeeType, TransactionType 1a
9from polar.postgres import AsyncSession 1a
11from .base import BaseTransactionService, BaseTransactionServiceError 1a
14class ProcessorFeeTransactionError(BaseTransactionServiceError): ... 1a
17class BalanceTransactionNotFound(ProcessorFeeTransactionError): 1a
18 def __init__(self, payment_transaction: Transaction) -> None: 1a
19 message = (
20 f"Balance transaction not found for payment transaction "
21 f"{payment_transaction.id} with charge ID {payment_transaction.charge_id}"
22 )
23 super().__init__(message)
26class UnsupportedStripeFeeType(ProcessorFeeTransactionError): 1a
27 def __init__(self, description: str) -> None: 1a
28 self.description = description
29 message = f"Unsupported Stripe fee type: {description}"
30 super().__init__(message)
33def _get_stripe_processor_fee_type(description: str) -> ProcessorFeeType: 1a
34 description = description.lower()
35 if "payout fee" in description or "account volume" in description:
36 return ProcessorFeeType.payout
37 if "cross-border transfers" in description:
38 return ProcessorFeeType.cross_border_transfer
39 if "active account" in description:
40 return ProcessorFeeType.account
41 if "billing" in description:
42 return ProcessorFeeType.subscription
43 if (
44 "automatic tax" in description
45 or "tax api calculation" in description
46 or "tax api transaction" in description
47 ):
48 return ProcessorFeeType.tax
49 if "invoicing" in description or "post payment invoices" in description:
50 return ProcessorFeeType.invoice
51 # Instant Bank Account Validation for ACH payments
52 if "connections verification" in description:
53 return ProcessorFeeType.payment
54 if "radar" in description:
55 return ProcessorFeeType.security
56 if "3d secure" in description:
57 return ProcessorFeeType.payment
58 if "authorization optimization" in description:
59 return ProcessorFeeType.payment
60 if "card account updater" in description:
61 return ProcessorFeeType.payment
62 if "tax reporting for connect" in description:
63 return ProcessorFeeType.tax
64 if "identity document check" in description:
65 return ProcessorFeeType.security
66 if "payments" in description:
67 return ProcessorFeeType.payment
68 if "card dispute countered fee" in description:
69 return ProcessorFeeType.dispute
70 raise UnsupportedStripeFeeType(description)
73class ProcessorFeeTransactionService(BaseTransactionService): 1a
74 async def create_payment_fees( 1a
75 self, session: AsyncSession, *, payment_transaction: Transaction
76 ) -> list[Transaction]:
77 fee_transactions: list[Transaction] = []
79 if payment_transaction.processor != Processor.stripe:
80 return fee_transactions
82 if payment_transaction.charge_id is None:
83 return fee_transactions
85 charge = await stripe_service.get_charge(payment_transaction.charge_id)
87 if charge.balance_transaction is None:
88 raise BalanceTransactionNotFound(payment_transaction)
90 stripe_balance_transaction = await stripe_service.get_balance_transaction(
91 get_expandable_id(charge.balance_transaction)
92 )
93 payment_fee_transaction = Transaction(
94 type=TransactionType.processor_fee,
95 processor=Processor.stripe,
96 processor_fee_type=ProcessorFeeType.payment,
97 currency=payment_transaction.currency,
98 amount=-stripe_balance_transaction.fee,
99 account_currency=payment_transaction.currency,
100 account_amount=-stripe_balance_transaction.fee,
101 tax_amount=0,
102 incurred_by_transaction=payment_transaction,
103 )
104 session.add(payment_fee_transaction)
105 fee_transactions.append(payment_fee_transaction)
107 await session.flush()
109 return fee_transactions
111 async def create_refund_fees( 1a
112 self,
113 session: AsyncSession,
114 *,
115 refund: Refund,
116 refund_transaction: Transaction,
117 ) -> list[Transaction]:
118 fee_transactions: list[Transaction] = []
120 is_stripe_refund = refund.processor == PaymentProcessor.stripe
121 is_stripe_refund_trx = refund_transaction.processor == Processor.stripe
122 if not (is_stripe_refund and is_stripe_refund_trx):
123 return fee_transactions
125 if refund_transaction.refund_id is None:
126 return fee_transactions
128 if refund.processor_balance_transaction_id is None:
129 return fee_transactions
131 balance_transaction = await stripe_service.get_balance_transaction(
132 refund.processor_balance_transaction_id,
133 )
135 refund_fee_transaction = Transaction(
136 type=TransactionType.processor_fee,
137 processor=refund_transaction.processor,
138 processor_fee_type=ProcessorFeeType.refund,
139 currency=refund_transaction.currency,
140 amount=-balance_transaction.fee,
141 account_currency=refund_transaction.currency,
142 account_amount=-balance_transaction.fee,
143 tax_amount=0,
144 incurred_by_transaction_id=refund_transaction.id,
145 )
147 session.add(refund_fee_transaction)
148 fee_transactions.append(refund_fee_transaction)
150 await session.flush()
152 return fee_transactions
154 async def create_dispute_fees( 1a
155 self,
156 session: AsyncSession,
157 *,
158 dispute_transaction: Transaction,
159 category: Literal["dispute", "dispute_reversal"],
160 ) -> list[Transaction]:
161 fee_transactions: list[Transaction] = []
163 if dispute_transaction.processor != Processor.stripe:
164 return fee_transactions
166 if dispute_transaction.dispute_id is None:
167 return fee_transactions
169 dispute = await stripe_service.get_dispute(dispute_transaction.dispute_id)
170 for balance_transaction in dispute.balance_transactions:
171 if (
172 balance_transaction.reporting_category == category
173 and balance_transaction.fee > 0
174 ):
175 dispute_fee_transaction = Transaction(
176 type=TransactionType.processor_fee,
177 processor=Processor.stripe,
178 processor_fee_type=ProcessorFeeType.dispute,
179 currency=dispute_transaction.currency,
180 amount=-balance_transaction.fee,
181 account_currency=dispute_transaction.currency,
182 account_amount=-balance_transaction.fee,
183 tax_amount=0,
184 incurred_by_transaction_id=dispute_transaction.id,
185 )
186 session.add(dispute_fee_transaction)
187 fee_transactions.append(dispute_fee_transaction)
189 await session.flush()
191 return fee_transactions
193 async def sync_stripe_fees(self, session: AsyncSession) -> list[Transaction]: 1a
194 transactions: list[Transaction] = []
196 balance_transactions = await stripe_service.list_balance_transactions(
197 type="stripe_fee", expand=["data.source"]
198 )
199 async for balance_transaction in balance_transactions:
200 transaction = await self.get_by(
201 session, fee_balance_transaction_id=balance_transaction.id
202 )
204 # We reached the point where we have already synced all the fees
205 if transaction is not None:
206 break
208 if balance_transaction.description is None:
209 continue
211 processor_fee_type = _get_stripe_processor_fee_type(
212 balance_transaction.description
213 )
214 transaction = Transaction(
215 created_at=datetime.fromtimestamp(balance_transaction.created, tz=UTC),
216 type=TransactionType.processor_fee,
217 processor=Processor.stripe,
218 processor_fee_type=processor_fee_type,
219 currency=balance_transaction.currency,
220 amount=balance_transaction.net,
221 account_currency=balance_transaction.currency,
222 account_amount=balance_transaction.net,
223 tax_amount=0,
224 fee_balance_transaction_id=balance_transaction.id,
225 )
226 session.add(transaction)
227 transactions.append(transaction)
229 await session.flush()
231 return transactions
234processor_fee_transaction = ProcessorFeeTransactionService(Transaction) 1a