Coverage for polar/invoice/service.py: 30%
63 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 15:52 +0000
1from datetime import datetime 1a
3from polar.config import settings 1a
4from polar.exceptions import PolarError 1a
5from polar.integrations.aws.s3 import S3Service 1a
6from polar.kit.tax import TaxabilityReason 1a
7from polar.kit.utils import utc_now 1a
8from polar.models import Account, Order, Payout 1a
9from polar.models.transaction import PlatformFeeType 1a
10from polar.postgres import AsyncSession 1a
11from polar.transaction.repository import TransactionRepository 1a
13from .generator import ( 1a
14 Invoice,
15 InvoiceGenerator,
16 InvoiceHeadingItem,
17 InvoiceItem,
18)
21class InvoiceError(PolarError): ... 1a
24class MissingAccountBillingDetails(InvoiceError): 1a
25 def __init__(self, account: Account) -> None: 1a
26 self.account = account
27 message = (
28 "You must provide billing details for the account to generate an invoice."
29 )
30 super().__init__(message, 400)
33class InvoiceService: 1a
34 async def create_order_invoice(self, order: Order) -> str: 1a
35 invoice = Invoice.from_order(order)
36 generator = InvoiceGenerator(invoice)
37 generator.generate()
38 invoice_bytes = generator.output()
40 s3 = S3Service(settings.S3_CUSTOMER_INVOICES_BUCKET_NAME)
41 return s3.upload(
42 bytes(invoice_bytes), order.invoice_filename, "application/pdf"
43 )
45 async def get_order_invoice_url(self, order: Order) -> tuple[str, datetime]: 1a
46 invoice_path = order.invoice_path
47 assert invoice_path is not None
48 s3 = S3Service(settings.S3_CUSTOMER_INVOICES_BUCKET_NAME)
49 return s3.generate_presigned_download_url(
50 path=invoice_path,
51 filename=order.invoice_filename,
52 mime_type="application/pdf",
53 )
55 async def create_payout_invoice(self, session: AsyncSession, payout: Payout) -> str: 1a
56 account = payout.account
57 if account.billing_name is None or account.billing_address is None:
58 raise MissingAccountBillingDetails(account)
60 transaction_repository = TransactionRepository.from_session(session)
61 payout_transactions = (
62 await transaction_repository.get_all_paid_transactions_by_payout(
63 payout.transaction.id
64 )
65 )
66 earliest = payout_transactions[0].created_at
67 latest = payout_transactions[-1].created_at
69 gross_amount = 0
70 payment_fees_amount = 0
71 payout_fees_amount = 0
72 for transaction in payout_transactions:
73 if transaction.platform_fee_type is None:
74 gross_amount += transaction.amount
75 elif transaction.platform_fee_type not in {
76 PlatformFeeType.account,
77 PlatformFeeType.payout,
78 PlatformFeeType.cross_border_transfer,
79 }:
80 payment_fees_amount += transaction.amount
81 else:
82 payout_fees_amount += transaction.amount
84 # Sanity check to make sure the amounts add up correctly
85 assert payout.fees_amount == abs(payout_fees_amount)
86 assert payout.amount == gross_amount + payout_fees_amount + payment_fees_amount
87 assert payout.paid_at is not None
89 invoice = Invoice(
90 number=payout.invoice_number,
91 date=utc_now(),
92 seller_name=account.billing_name,
93 seller_address=account.billing_address,
94 seller_additional_info=account.billing_additional_info,
95 customer_name=settings.INVOICES_NAME,
96 customer_address=settings.INVOICES_ADDRESS,
97 customer_additional_info=settings.INVOICES_ADDITIONAL_INFO,
98 subtotal_amount=payout.amount,
99 discount_amount=0,
100 taxability_reason=TaxabilityReason.product_exempt,
101 tax_amount=0,
102 tax_rate=None,
103 currency=payout.currency,
104 items=[
105 InvoiceItem(
106 description=f"Digital services and products resold by Polar.sh\nFrom {earliest.strftime('%Y-%m-%d')} to {latest.strftime('%Y-%m-%d')}",
107 quantity=1,
108 unit_amount=gross_amount,
109 amount=gross_amount,
110 ),
111 InvoiceItem(
112 description="Polar revenue share",
113 quantity=1,
114 unit_amount=payment_fees_amount,
115 amount=payment_fees_amount,
116 ),
117 InvoiceItem(
118 description="Payout fees",
119 quantity=1,
120 unit_amount=payout_fees_amount,
121 amount=payout_fees_amount,
122 ),
123 ],
124 notes=(f"{account.billing_notes}\n\n" if account.billing_notes else "")
125 + (
126 "Polar Software, Inc. is the merchant of record reselling digital services.\n"
127 "Polar Software, Inc. captures and remits international sales tax from such sales – as needed.\n"
128 "Payouts (reverse invoices) are therefore without taxes."
129 ),
130 extra_heading_items=[
131 InvoiceHeadingItem(label="Paid at", value=payout.paid_at),
132 InvoiceHeadingItem(
133 label="Payout Method", value=payout.processor.get_display_name()
134 ),
135 InvoiceHeadingItem(label="Payout ID", value=str(payout.id)),
136 ],
137 )
139 generator = InvoiceGenerator(invoice, heading_title="Reverse Invoice")
140 generator.generate()
141 invoice_bytes = generator.output()
142 s3 = S3Service(settings.S3_PAYOUT_INVOICES_BUCKET_NAME)
143 return s3.upload(
144 bytes(invoice_bytes),
145 f"{account.id}/Payout-{payout.invoice_number}.pdf",
146 "application/pdf",
147 )
149 async def get_payout_invoice_url(self, payout: Payout) -> tuple[str, datetime]: 1a
150 invoice_path = payout.invoice_path
151 assert invoice_path is not None
152 filename = f"Payout-{payout.invoice_number}.pdf"
153 s3 = S3Service(settings.S3_PAYOUT_INVOICES_BUCKET_NAME)
154 return s3.generate_presigned_download_url(
155 path=invoice_path, filename=filename, mime_type="application/pdf"
156 )
159invoice = InvoiceService() 1a