Coverage for polar/transaction/service/payment.py: 31%
50 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 16:17 +0000
1from typing import cast 1a
2from uuid import UUID 1a
4import stripe as stripe_lib 1a
5from sqlalchemy import select 1a
7from polar.integrations.stripe.schemas import ProductType 1a
8from polar.integrations.stripe.service import stripe as stripe_service 1a
9from polar.integrations.stripe.utils import get_expandable_id 1a
10from polar.models import Pledge, Transaction 1a
11from polar.models.transaction import Processor, TransactionType 1a
12from polar.postgres import AsyncSession 1a
13from polar.worker import enqueue_job 1a
15from .base import BaseTransactionService, BaseTransactionServiceError 1a
18class PaymentTransactionError(BaseTransactionServiceError): ... 1a
21class PaymentTransactionService(BaseTransactionService): 1a
22 async def get_by_charge_id( 1a
23 self, session: AsyncSession, charge_id: str
24 ) -> Transaction | None:
25 statement = select(Transaction).where(
26 Transaction.type == TransactionType.payment,
27 Transaction.charge_id == charge_id,
28 )
29 result = await session.execute(statement)
30 return result.scalar_one_or_none()
32 async def get_by_order_id( 1a
33 self, session: AsyncSession, order_id: UUID
34 ) -> Transaction | None:
35 return await self.get_by(
36 session,
37 type=TransactionType.payment,
38 order_id=order_id,
39 )
41 async def create_payment( 1a
42 self, session: AsyncSession, *, charge: stripe_lib.Charge
43 ) -> Transaction:
44 pledge: Pledge | None = None
46 # Make sure we don't already have this transaction
47 existing_transaction = await self.get_by_charge_id(session, charge.id)
48 if existing_transaction is not None:
49 return existing_transaction
51 # Retrieve tax amount and country
52 tax_amount = 0
53 tax_country = None
54 tax_state = None
55 pledge_invoice = False
56 # Polar Custom Checkout sets tax info in metadata
57 if "tax_amount" in charge.metadata:
58 tax_amount = int(charge.metadata["tax_amount"])
59 tax_country = charge.metadata["tax_country"]
60 tax_state = charge.metadata.get("tax_state")
61 # Stripe Checkout sets tax info in invoice
62 elif charge.invoice:
63 stripe_invoice = await stripe_service.get_invoice(
64 get_expandable_id(charge.invoice)
65 )
66 if stripe_invoice.tax is not None:
67 tax_amount = stripe_invoice.tax
68 for total_tax_amount in stripe_invoice.total_tax_amounts:
69 tax_rate = cast(stripe_lib.TaxRate, total_tax_amount.tax_rate)
70 tax_country = tax_rate.country
71 tax_state = tax_rate.state
73 if (
74 stripe_invoice.metadata
75 and stripe_invoice.metadata.get("type") == ProductType.pledge
76 ):
77 pledge_invoice = True
79 risk = getattr(charge, "outcome", {})
80 transaction = Transaction(
81 type=TransactionType.payment,
82 processor=Processor.stripe,
83 currency=charge.currency,
84 amount=charge.amount - tax_amount,
85 account_currency=charge.currency,
86 account_amount=charge.amount - tax_amount,
87 tax_amount=tax_amount,
88 tax_country=tax_country,
89 tax_state=tax_state if tax_country in {"US", "CA"} else None,
90 presentment_currency=charge.currency,
91 presentment_amount=charge.amount - tax_amount,
92 presentment_tax_amount=tax_amount,
93 customer_id=get_expandable_id(charge.customer) if charge.customer else None,
94 charge_id=charge.id,
95 pledge=pledge,
96 risk_level=risk.get("risk_level"),
97 risk_score=risk.get("risk_score"),
98 # Filled when we handle the invoice
99 order=None,
100 payment_customer=None,
101 # Legacy fields for pledges
102 payment_organization=None,
103 payment_user=None,
104 )
106 session.add(transaction)
107 await session.flush()
109 # Enqueue fees creation
110 enqueue_job("processor_fee.create_payment_fees", transaction.id)
112 return transaction
115payment_transaction = PaymentTransactionService(Transaction) 1a