Coverage for polar/kit/tax.py: 77%
285 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 12:42 +0000
1import hashlib
2import json
3import uuid
4from collections.abc import Sequence
5from enum import StrEnum
6from typing import Annotated, Any, Literal, LiteralString, Protocol, TypedDict
8import stdnum.ca.bn
9import stdnum.cl.rut
10import stdnum.exceptions
11import stdnum.in_.gstin
12import stdnum.tr.vkn
13import stripe as stripe_lib
14import structlog
15from pydantic import Field
16from sqlalchemy.dialects.postgresql import JSONB
17from sqlalchemy.engine.interfaces import Dialect
18from sqlalchemy.types import TypeDecorator
19from stdnum import get_cc_module
21from polar.config import settings
22from polar.exceptions import PolarError
23from polar.integrations.stripe.service import stripe as stripe_service
24from polar.kit.address import Address
25from polar.logging import Logger
27log: Logger = structlog.get_logger()
30class TaxIDFormat(StrEnum):
31 """
32 List of supported tax ID formats.
34 Ref: https://docs.stripe.com/billing/customer/tax-ids#supported-tax-id
35 """
37 ad_nrt = "ad_nrt"
38 ae_trn = "ae_trn"
39 ar_cuit = "ar_cuit"
40 au_abn = "au_abn"
41 au_arn = "au_arn"
42 bg_uic = "bg_uic"
43 bh_vat = "bh_vat"
44 bo_tin = "bo_tin"
45 br_cnpj = "br_cnpj"
46 br_cpf = "br_cpf"
47 ca_bn = "ca_bn"
48 ca_gst_hst = "ca_gst_hst"
49 ca_pst_bc = "ca_pst_bc"
50 ca_pst_mb = "ca_pst_mb"
51 ca_pst_sk = "ca_pst_sk"
52 ca_qst = "ca_qst"
53 ch_uid = "ch_uid"
54 ch_vat = "ch_vat"
55 cl_tin = "cl_tin"
56 cn_tin = "cn_tin"
57 co_nit = "co_nit"
58 cr_tin = "cr_tin"
59 de_stn = "de_stn"
60 do_rcn = "do_rcn"
61 ec_ruc = "ec_ruc"
62 eg_tin = "eg_tin"
63 es_cif = "es_cif"
64 eu_oss_vat = "eu_oss_vat"
65 eu_vat = "eu_vat"
66 gb_vat = "gb_vat"
67 ge_vat = "ge_vat"
68 hk_br = "hk_br"
69 hr_oib = "hr_oib"
70 hu_tin = "hu_tin"
71 id_npwp = "id_npwp"
72 il_vat = "il_vat"
73 in_gst = "in_gst"
74 is_vat = "is_vat"
75 jp_cn = "jp_cn"
76 jp_rn = "jp_rn"
77 jp_trn = "jp_trn"
78 ke_pin = "ke_pin"
79 kr_brn = "kr_brn"
80 kz_bin = "kz_bin"
81 li_uid = "li_uid"
82 mx_rfc = "mx_rfc"
83 my_frp = "my_frp"
84 my_itn = "my_itn"
85 my_sst = "my_sst"
86 ng_tin = "ng_tin"
87 no_vat = "no_vat"
88 no_voec = "no_voec"
89 nz_gst = "nz_gst"
90 om_vat = "om_vat"
91 pe_ruc = "pe_ruc"
92 ph_tin = "ph_tin"
93 ro_tin = "ro_tin"
94 rs_pib = "rs_pib"
95 ru_inn = "ru_inn"
96 ru_kpp = "ru_kpp"
97 sa_vat = "sa_vat"
98 sg_gst = "sg_gst"
99 sg_uen = "sg_uen"
100 si_tin = "si_tin"
101 sv_nit = "sv_nit"
102 th_vat = "th_vat"
103 tr_tin = "tr_tin"
104 tw_vat = "tw_vat"
105 ua_vat = "ua_vat"
106 us_ein = "us_ein"
107 uy_ruc = "uy_ruc"
108 ve_rif = "ve_rif"
109 vn_tin = "vn_tin"
110 za_vat = "za_vat"
113COUNTRY_TAX_ID_MAP: dict[str, Sequence[TaxIDFormat]] = {
114 "AD": (TaxIDFormat.ad_nrt,),
115 "AE": (TaxIDFormat.ae_trn,),
116 "AR": (TaxIDFormat.ar_cuit,),
117 "AT": (TaxIDFormat.eu_vat,),
118 "AU": (TaxIDFormat.au_abn, TaxIDFormat.au_arn),
119 "BE": (TaxIDFormat.eu_vat,),
120 "BG": (TaxIDFormat.bg_uic, TaxIDFormat.eu_vat),
121 "BH": (TaxIDFormat.bh_vat,),
122 "BO": (TaxIDFormat.bo_tin,),
123 "BR": (TaxIDFormat.br_cnpj, TaxIDFormat.br_cpf),
124 "CA": (
125 TaxIDFormat.ca_gst_hst,
126 TaxIDFormat.ca_pst_bc,
127 TaxIDFormat.ca_pst_mb,
128 TaxIDFormat.ca_pst_sk,
129 TaxIDFormat.ca_qst,
130 TaxIDFormat.ca_bn,
131 ),
132 "CH": (TaxIDFormat.ch_uid, TaxIDFormat.ch_vat),
133 "CL": (TaxIDFormat.cl_tin,),
134 "CN": (TaxIDFormat.cn_tin,),
135 "CO": (TaxIDFormat.co_nit,),
136 "CR": (TaxIDFormat.cr_tin,),
137 "CY": (TaxIDFormat.eu_vat,),
138 "CZ": (TaxIDFormat.eu_vat,),
139 "DE": (TaxIDFormat.de_stn, TaxIDFormat.eu_vat),
140 "DK": (TaxIDFormat.eu_vat,),
141 "DO": (TaxIDFormat.do_rcn,),
142 "EC": (TaxIDFormat.ec_ruc,),
143 "EE": (TaxIDFormat.eu_vat,),
144 "EG": (TaxIDFormat.eg_tin,),
145 "ES": (TaxIDFormat.es_cif, TaxIDFormat.eu_vat),
146 "FI": (TaxIDFormat.eu_vat,),
147 "FR": (TaxIDFormat.eu_vat,),
148 "GB": (TaxIDFormat.gb_vat,),
149 "GE": (TaxIDFormat.ge_vat,),
150 "GR": (TaxIDFormat.eu_vat,),
151 "HK": (TaxIDFormat.hk_br,),
152 "HR": (TaxIDFormat.hr_oib, TaxIDFormat.eu_vat),
153 "HU": (TaxIDFormat.hu_tin, TaxIDFormat.eu_vat),
154 "ID": (TaxIDFormat.id_npwp,),
155 "IE": (TaxIDFormat.eu_vat,),
156 "IL": (TaxIDFormat.il_vat,),
157 "IN": (TaxIDFormat.in_gst,),
158 "IS": (TaxIDFormat.is_vat,),
159 "IT": (TaxIDFormat.eu_vat,),
160 "JP": (TaxIDFormat.jp_cn, TaxIDFormat.jp_rn, TaxIDFormat.jp_trn),
161 "KE": (TaxIDFormat.ke_pin,),
162 "KR": (TaxIDFormat.kr_brn,),
163 "KZ": (TaxIDFormat.kz_bin,),
164 "LI": (TaxIDFormat.li_uid,),
165 "LT": (TaxIDFormat.eu_vat,),
166 "LU": (TaxIDFormat.eu_vat,),
167 "LV": (TaxIDFormat.eu_vat,),
168 "MT": (TaxIDFormat.eu_vat,),
169 "MX": (TaxIDFormat.mx_rfc,),
170 "MY": (TaxIDFormat.my_frp, TaxIDFormat.my_itn, TaxIDFormat.my_sst),
171 "NG": (TaxIDFormat.ng_tin,),
172 "NL": (TaxIDFormat.eu_vat,),
173 "NO": (TaxIDFormat.no_vat, TaxIDFormat.no_voec),
174 "NZ": (TaxIDFormat.nz_gst,),
175 "OM": (TaxIDFormat.om_vat,),
176 "PE": (TaxIDFormat.pe_ruc,),
177 "PH": (TaxIDFormat.ph_tin,),
178 "PL": (TaxIDFormat.eu_vat,),
179 "PT": (TaxIDFormat.eu_vat,),
180 "RO": (TaxIDFormat.ro_tin, TaxIDFormat.eu_vat),
181 "RS": (TaxIDFormat.rs_pib,),
182 "RU": (TaxIDFormat.ru_inn, TaxIDFormat.ru_kpp),
183 "SA": (TaxIDFormat.sa_vat,),
184 "SE": (TaxIDFormat.eu_vat,),
185 "SG": (TaxIDFormat.sg_gst, TaxIDFormat.sg_uen),
186 "SI": (TaxIDFormat.si_tin, TaxIDFormat.eu_vat),
187 "SK": (TaxIDFormat.eu_vat,),
188 "SV": (TaxIDFormat.sv_nit,),
189 "TH": (TaxIDFormat.th_vat,),
190 "TR": (TaxIDFormat.tr_tin,),
191 "TW": (TaxIDFormat.tw_vat,),
192 "UA": (TaxIDFormat.ua_vat,),
193 "US": (TaxIDFormat.us_ein,),
194 "UY": (TaxIDFormat.uy_ruc,),
195 "VE": (TaxIDFormat.ve_rif,),
196 "VN": (TaxIDFormat.vn_tin,),
197 "ZA": (TaxIDFormat.za_vat,),
198}
200TaxID = Annotated[
201 tuple[str, TaxIDFormat],
202 Field(examples=[("911144442", "us_ein"), ("FR61954506077", "eu_vat")]),
203]
206class TaxError(PolarError): ... 206 ↛ 209line 206 didn't jump to line 209 because
209class UnsupportedTaxIDFormat(TaxError):
210 def __init__(self, tax_id_type: TaxIDFormat) -> None:
211 self.tax_id_type = tax_id_type
212 super().__init__(f"Tax ID format {tax_id_type} is not supported.")
215class InvalidTaxID(TaxError):
216 def __init__(self, tax_id: str, country: str) -> None:
217 self.tax_id = tax_id
218 self.country = country
219 super().__init__("Invalid tax ID.")
222class ValidatorProtocol(Protocol):
223 def validate(self, number: str, country: str) -> str: ... 223 ↛ exitline 223 didn't return from function 'validate' because
226class StdNumValidator(ValidatorProtocol):
227 def __init__(self, tax_id_type: TaxIDFormat):
228 tax_id_country, tax_id_format = tax_id_type.split("_", 1)
229 module = get_cc_module(tax_id_country, tax_id_format)
230 if module is None:
231 raise UnsupportedTaxIDFormat(tax_id_type)
232 self.module = module
234 def validate(self, number: str, country: str) -> str:
235 try:
236 return self.module.validate(number)
237 except stdnum.exceptions.ValidationError as e:
238 raise InvalidTaxID(number, country) from e
241class CAGSTHSTValidator(ValidatorProtocol):
242 def validate(self, number: str, country: str) -> str:
243 number = stdnum.ca.bn.compact(number)
244 if len(number) != 15: 244 ↛ 246line 244 didn't jump to line 246 because the condition on line 244 was always true
245 raise InvalidTaxID(number, country)
246 try:
247 return stdnum.ca.bn.validate(number)
248 except stdnum.exceptions.ValidationError as e:
249 raise InvalidTaxID(number, country) from e
252class CLTINValidator(ValidatorProtocol):
253 def validate(self, number: str, country: str) -> str:
254 number = stdnum.cl.rut.compact(number)
255 try:
256 return stdnum.cl.rut.validate(number)
257 except stdnum.exceptions.ValidationError as e:
258 raise InvalidTaxID(number, country) from e
261class TRTINValidator(ValidatorProtocol):
262 def validate(self, number: str, country: str) -> str:
263 number = stdnum.tr.vkn.compact(number)
264 try:
265 return stdnum.tr.vkn.validate(number)
266 except stdnum.exceptions.ValidationError as e:
267 raise InvalidTaxID(number, country) from e
270class INGSTValidator(ValidatorProtocol):
271 def validate(self, number: str, country: str) -> str:
272 number = stdnum.in_.gstin.compact(number)
273 try:
274 return stdnum.in_.gstin.validate(number)
275 except stdnum.exceptions.ValidationError as e:
276 raise InvalidTaxID(number, country) from e
279def _get_validator(tax_id_type: TaxIDFormat) -> ValidatorProtocol:
280 match tax_id_type:
281 case TaxIDFormat.ca_gst_hst:
282 return CAGSTHSTValidator()
283 case TaxIDFormat.cl_tin:
284 return CLTINValidator()
285 case TaxIDFormat.tr_tin:
286 return TRTINValidator()
287 case TaxIDFormat.in_gst:
288 return INGSTValidator()
289 case _:
290 return StdNumValidator(tax_id_type)
293def validate_tax_id(number: str, country: str) -> TaxID:
294 """
295 Validate a tax ID for a given country.
297 Args:
298 number: The tax ID to validate.
299 country: The country of the tax ID.
301 Returns:
302 The validated tax ID and the tax ID format as tuple
304 Raises:
305 InvalidTaxID: The tax ID is invalid or unsupported.
306 """
307 try:
308 tax_id_types = COUNTRY_TAX_ID_MAP[country]
309 except KeyError as e:
310 raise InvalidTaxID(number, country) from e
311 else:
312 for tax_id_type in tax_id_types:
313 try:
314 validator = _get_validator(tax_id_type)
315 return validator.validate(number, country), tax_id_type
316 except (UnsupportedTaxIDFormat, InvalidTaxID):
317 continue
318 raise InvalidTaxID(number, country)
321def to_stripe_tax_id(value: TaxID) -> stripe_lib.Customer.CreateParamsTaxIdDatum:
322 """
323 Convert a tax ID to the format expected by Stripe.
325 Args:
326 value: A tuple containing the tax ID and the tax ID type.
328 Returns:
329 A dictionary containing the tax ID in the format expected by Stripe.
330 """
331 tax_id, tax_id_type = value
332 return {
333 "type": str(tax_id_type), # type: ignore
334 "value": tax_id,
335 }
338class TaxIDType(TypeDecorator[Any]):
339 impl = JSONB
340 cache_ok = True
342 def process_bind_param(self, value: Any, dialect: Dialect) -> Any:
343 if value is not None:
344 if not isinstance(value, tuple | list) or len(value) != 2: 344 ↛ 345line 344 didn't jump to line 345 because the condition on line 344 was never true
345 raise TypeError("Invalid tax ID value.")
346 return json.dumps(value)
347 return value
349 def process_result_value(self, value: str | None, dialect: Dialect) -> Any:
350 if value is not None:
351 return json.loads(value)
352 return value
355class TaxCalculationError(PolarError):
356 message: LiteralString
358 def __init__(
359 self,
360 stripe_error: stripe_lib.StripeError,
361 message: LiteralString = "An error occurred while calculating tax.",
362 ) -> None:
363 self.stripe_error = stripe_error
364 self.message = message
365 super().__init__(message)
368class IncompleteTaxLocation(TaxCalculationError):
369 def __init__(self, stripe_error: stripe_lib.InvalidRequestError) -> None:
370 super().__init__(stripe_error, "Required tax location information is missing.")
373class InvalidTaxLocation(TaxCalculationError):
374 def __init__(self, stripe_error: stripe_lib.StripeError) -> None:
375 super().__init__(
376 stripe_error,
377 (
378 "We could not determine the customer's tax location "
379 "based on the provided customer address."
380 ),
381 )
384class TaxabilityReason(StrEnum):
385 standard_rated = "standard_rated"
386 """Purchases that are subject to the standard rate of tax."""
388 not_collecting = "not_collecting"
389 """Purchases for countries where we don't collect tax."""
391 product_exempt = "product_exempt"
392 """Purchases for products that are exempt from tax."""
394 reverse_charge = "reverse_charge"
395 """Purchases where the customer is responsible for paying tax, e.g. B2B transactions with provided tax ID."""
397 not_subject_to_tax = "not_subject_to_tax"
398 """Purchases where the customer provided a tax ID, but on countries where we don't collect tax."""
400 not_supported = "not_supported"
401 """Purchases from countries where we don't support tax."""
403 customer_exempt = "customer_exempt"
404 """Purchases where the customer is exempt from tax, e.g. if the subscription was created before our tax registration."""
406 @classmethod
407 def from_stripe(
408 cls, stripe_reason: str | None, tax_amount: int
409 ) -> "TaxabilityReason | None":
410 if stripe_reason is None or stripe_reason == "not_available":
411 # Stripe sometimes returns `None` or `not_available` even if taxes are collected.
412 if tax_amount != 0:
413 return TaxabilityReason.standard_rated
414 return None
416 return cls(stripe_reason)
419class TaxRate(TypedDict):
420 rate_type: Literal["percentage"] | Literal["fixed"]
421 basis_points: int | None
422 amount: int | None
423 amount_currency: str | None
424 display_name: str
425 country: str | None
426 state: str | None
429def from_stripe_tax_rate(tax_rate: stripe_lib.TaxRate) -> TaxRate | None:
430 rate_type = tax_rate.rate_type
431 if rate_type is None:
432 return None
434 return {
435 "rate_type": "fixed" if rate_type == "flat_amount" else "percentage",
436 "basis_points": int(tax_rate.percentage * 100)
437 if tax_rate.percentage is not None
438 else None,
439 "amount": tax_rate.flat_amount.amount if tax_rate.flat_amount else None,
440 "amount_currency": tax_rate.flat_amount.currency
441 if tax_rate.flat_amount
442 else None,
443 "display_name": tax_rate.display_name,
444 "country": tax_rate.country,
445 "state": tax_rate.state,
446 }
449def from_stripe_tax_rate_details(
450 tax_rate_details: stripe_lib.tax.Calculation.TaxBreakdown.TaxRateDetails,
451) -> TaxRate | None:
452 rate_type = tax_rate_details.rate_type
453 if rate_type is None:
454 return None
456 basis_points = None
457 amount = None
458 amount_currency = None
460 if tax_rate_details.percentage_decimal is not None:
461 basis_points = int(float(tax_rate_details.percentage_decimal) * 100)
462 elif tax_rate_details.flat_amount is not None:
463 amount = tax_rate_details.flat_amount.amount
464 amount_currency = tax_rate_details.flat_amount.currency
466 tax_type = tax_rate_details.tax_type
467 display_name = "Tax"
468 if tax_type is not None:
469 if tax_type in {"gst", "hst", "igst", "jct", "pst", "qct", "rst", "vat"}:
470 display_name = tax_type.upper()
471 else:
472 display_name = tax_type.replace("_", " ").title()
474 return {
475 "rate_type": "fixed" if rate_type == "flat_amount" else "percentage",
476 "basis_points": basis_points,
477 "amount": amount,
478 "amount_currency": amount_currency,
479 "display_name": display_name,
480 "country": tax_rate_details.country,
481 "state": tax_rate_details.state,
482 }
485class TaxCode(StrEnum):
486 general_electronically_supplied_services = (
487 "general_electronically_supplied_services"
488 )
490 def to_stripe(self) -> str:
491 match self:
492 case TaxCode.general_electronically_supplied_services: 492 ↛ exitline 492 didn't return from function 'to_stripe' because the pattern on line 492 always matched
493 return "txcd_10000000"
496class TaxCalculation(TypedDict):
497 processor_id: str
498 amount: int
499 taxability_reason: TaxabilityReason | None
500 tax_rate: TaxRate | None
503async def calculate_tax(
504 identifier: uuid.UUID | str,
505 currency: str,
506 amount: int,
507 tax_code: TaxCode,
508 address: Address,
509 tax_ids: list[TaxID],
510 customer_exempt: bool,
511) -> TaxCalculation:
512 # Compute an idempotency key based on the input parameters to work as a sort of cache
513 address_str = address.model_dump_json()
514 tax_ids_str = ",".join(f"{tax_id[0]}:{tax_id[1]}" for tax_id in tax_ids)
515 taxability_override: Literal["customer_exempt", "none"] = (
516 "customer_exempt" if customer_exempt else "none"
517 )
518 idempotency_key_str = f"{identifier}:{currency}:{amount}:{tax_code}:{address_str}:{tax_ids_str}:{taxability_override}"
519 idempotency_key = hashlib.sha256(idempotency_key_str.encode()).hexdigest()
521 try:
522 calculation = await stripe_service.create_tax_calculation(
523 currency=currency,
524 line_items=[
525 {
526 "amount": amount,
527 "tax_code": tax_code.to_stripe(),
528 "quantity": 1,
529 "reference": str(identifier),
530 }
531 ],
532 customer_details={
533 "address": address.to_dict(),
534 "address_source": "billing",
535 "tax_ids": [to_stripe_tax_id(tax_id) for tax_id in tax_ids],
536 "taxability_override": taxability_override,
537 },
538 idempotency_key=idempotency_key,
539 )
540 except stripe_lib.RateLimitError as e:
541 if settings.is_sandbox():
542 log.warning(
543 "Stripe Tax API rate limit exceeded in sandbox mode, returning zero tax",
544 identifier=str(identifier),
545 currency=currency,
546 amount=amount,
547 )
548 return {
549 "processor_id": f"taxcalc_sandbox_{uuid.uuid4().hex}",
550 "amount": 0,
551 "taxability_reason": None,
552 "tax_rate": None,
553 }
554 raise
555 except stripe_lib.InvalidRequestError as e:
556 if (
557 e.error is not None
558 and e.error.param is not None
559 and e.error.param.startswith("customer_details[address]")
560 ):
561 raise IncompleteTaxLocation(e) from e
562 raise
563 except stripe_lib.StripeError as e:
564 if e.error is None or e.error.code != "customer_tax_location_invalid": 564 ↛ 566line 564 didn't jump to line 566 because the condition on line 564 was always true
565 raise
566 raise InvalidTaxLocation(e) from e
567 else:
568 assert calculation.id is not None
569 amount = calculation.tax_amount_exclusive
570 breakdown = calculation.tax_breakdown[0]
571 return {
572 "processor_id": calculation.id,
573 "amount": amount,
574 "taxability_reason": TaxabilityReason.from_stripe(
575 breakdown.taxability_reason, amount
576 ),
577 "tax_rate": from_stripe_tax_rate_details(breakdown.tax_rate_details),
578 }