Coverage for polar/product/schemas.py: 79%
203 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 17:15 +0000
1import builtins 1a
2from decimal import Decimal 1a
3from typing import Annotated, Any, Literal 1a
5from pydantic import UUID4, Discriminator, Field, Tag, computed_field, field_validator 1a
6from pydantic.aliases import AliasChoices 1a
7from pydantic.json_schema import SkipJsonSchema 1a
9from polar.benefit.schemas import Benefit, BenefitID, BenefitPublic 1a
10from polar.custom_field.schemas import ( 1a
11 AttachedCustomField,
12 AttachedCustomFieldListCreate,
13)
14from polar.enums import SubscriptionRecurringInterval 1a
15from polar.file.schemas import ProductMediaFileRead 1a
16from polar.kit.db.models import Model 1a
17from polar.kit.metadata import ( 1a
18 MetadataInputMixin,
19 MetadataOutputMixin,
20)
21from polar.kit.schemas import ( 1a
22 EmptyStrToNoneValidator,
23 IDSchema,
24 MergeJSONSchema,
25 Schema,
26 SelectorWidget,
27 SetSchemaReference,
28 TimestampedSchema,
29)
30from polar.kit.trial import TrialConfigurationInputMixin, TrialConfigurationOutputMixin 1a
31from polar.models.product_price import ( 1a
32 ProductPriceAmountType,
33 ProductPriceSource,
34 ProductPriceType,
35)
36from polar.models.product_price import ( 1a
37 ProductPriceCustom as ProductPriceCustomModel,
38)
39from polar.models.product_price import ( 1a
40 ProductPriceFixed as ProductPriceFixedModel,
41)
42from polar.models.product_price import ( 1a
43 ProductPriceFree as ProductPriceFreeModel,
44)
45from polar.models.product_price import ( 1a
46 ProductPriceMeteredUnit as ProductPriceMeteredUnitModel,
47)
48from polar.models.product_price import ( 1a
49 ProductPriceSeatUnit as ProductPriceSeatUnitModel,
50)
51from polar.organization.schemas import OrganizationID 1a
53PRODUCT_NAME_MIN_LENGTH = 3 1a
55# PostgreSQL int4 range limit
56INT_MAX_VALUE = 2_147_483_647 1a
58# Product
60ProductID = Annotated[ 1a
61 UUID4,
62 MergeJSONSchema({"description": "The product ID."}),
63 SelectorWidget("/v1/products", "Product", "name"),
64]
66# Ref: https://stripe.com/docs/api/payment_intents/object#payment_intent_object-amount
67MAXIMUM_PRICE_AMOUNT = 99999999 1a
68MINIMUM_PRICE_AMOUNT = 50 1a
71PriceAmount = Annotated[ 1a
72 int,
73 Field(
74 ...,
75 ge=MINIMUM_PRICE_AMOUNT,
76 le=MAXIMUM_PRICE_AMOUNT,
77 description="The price in cents.",
78 ),
79]
80SeatPriceAmount = Annotated[ 1a
81 int,
82 Field(
83 ...,
84 ge=0,
85 le=MAXIMUM_PRICE_AMOUNT,
86 description="The price per seat in cents. Can be 0 for free tiers.",
87 ),
88]
89PriceCurrency = Annotated[ 1a
90 str,
91 Field(
92 pattern="usd",
93 description="The currency. Currently, only `usd` is supported.",
94 ),
95]
96ProductName = Annotated[ 1a
97 str,
98 Field(
99 min_length=PRODUCT_NAME_MIN_LENGTH,
100 description="The name of the product.",
101 ),
102]
103ProductDescription = Annotated[ 1a
104 str | None,
105 Field(description="The description of the product."),
106 EmptyStrToNoneValidator,
107]
110class ProductPriceCreateBase(Schema): 1a
111 amount_type: ProductPriceAmountType 1a
113 def get_model_class(self) -> builtins.type[Model]: 1a
114 raise NotImplementedError()
117class ProductPriceFixedCreate(ProductPriceCreateBase): 1a
118 """
119 Schema to create a fixed price.
120 """
122 amount_type: Literal[ProductPriceAmountType.fixed] 1a
123 price_amount: PriceAmount 1a
124 price_currency: PriceCurrency = "usd" 1a
126 def get_model_class(self) -> builtins.type[ProductPriceFixedModel]: 1a
127 return ProductPriceFixedModel
130class ProductPriceCustomCreate(ProductPriceCreateBase): 1a
131 """
132 Schema to create a pay-what-you-want price.
133 """
135 amount_type: Literal[ProductPriceAmountType.custom] 1a
136 price_currency: PriceCurrency = "usd" 1a
137 minimum_amount: PriceAmount | None = Field( 1a
138 default=None, ge=50, description="The minimum amount the customer can pay."
139 )
140 maximum_amount: PriceAmount | None = Field( 1a
141 default=None,
142 le=1_000_000, # $10K
143 description="The maximum amount the customer can pay.",
144 )
145 preset_amount: PriceAmount | None = Field( 1a
146 default=None,
147 le=1_000_000, # $10K
148 description="The initial amount shown to the customer.",
149 )
151 def get_model_class(self) -> builtins.type[ProductPriceCustomModel]: 1a
152 return ProductPriceCustomModel
155class ProductPriceFreeCreate(ProductPriceCreateBase): 1a
156 """
157 Schema to create a free price.
158 """
160 amount_type: Literal[ProductPriceAmountType.free] 1a
162 def get_model_class(self) -> builtins.type[ProductPriceFreeModel]: 1a
163 return ProductPriceFreeModel
166class ProductPriceSeatTier(Schema): 1a
167 """
168 A pricing tier for seat-based pricing.
169 """
171 min_seats: int = Field(ge=1, description="Minimum number of seats (inclusive)") 1a
172 max_seats: int | None = Field( 1a
173 default=None,
174 ge=1,
175 description="Maximum number of seats (inclusive). None for unlimited.",
176 )
177 price_per_seat: SeatPriceAmount = Field( 1a
178 description="Price per seat in cents for this tier"
179 )
182class ProductPriceSeatTiers(Schema): 1a
183 """
184 List of pricing tiers for seat-based pricing.
185 """
187 tiers: list[ProductPriceSeatTier] = Field( 1a
188 min_length=1, description="List of pricing tiers"
189 )
191 @field_validator("tiers") 1a
192 @classmethod 1a
193 def validate_tiers( 1a
194 cls, v: list[ProductPriceSeatTier]
195 ) -> list[ProductPriceSeatTier]:
196 """Validate that tiers form continuous ranges without gaps or overlaps."""
197 if not v:
198 raise ValueError("At least one tier is required")
200 # Sort by min_seats
201 sorted_tiers = sorted(v, key=lambda t: t.min_seats)
203 # Ensure first tier starts at 1
204 if sorted_tiers[0].min_seats != 1:
205 raise ValueError("First tier must start at min_seats=1")
207 # Validate continuous ranges without gaps/overlaps
208 for i in range(len(sorted_tiers) - 1):
209 current = sorted_tiers[i]
210 next_tier = sorted_tiers[i + 1]
212 if current.max_seats is None:
213 raise ValueError(
214 "Only the last tier can have unlimited max_seats (None)"
215 )
217 if next_tier.min_seats != current.max_seats + 1:
218 raise ValueError(
219 "Gap or overlap between tiers: "
220 + f"tier ending at {current.max_seats} and tier starting at {next_tier.min_seats}"
221 )
223 # Ensure last tier has unlimited max_seats
224 last_tier = sorted_tiers[-1]
225 if last_tier.max_seats is not None:
226 raise ValueError(
227 "Last tier must have unlimited max_seats (None). "
228 + f"Currently set to {last_tier.max_seats}."
229 )
231 return sorted_tiers
234class ProductPriceSeatBasedCreate(ProductPriceCreateBase): 1a
235 """
236 Schema to create a seat-based price with volume-based tiers.
237 """
239 amount_type: Literal[ProductPriceAmountType.seat_based] 1a
240 price_currency: PriceCurrency = "usd" 1a
241 seat_tiers: ProductPriceSeatTiers = Field( 1a
242 description="Tiered pricing based on seat quantity"
243 )
245 def get_model_class(self) -> builtins.type[ProductPriceSeatUnitModel]: 1a
246 return ProductPriceSeatUnitModel
249class ProductPriceMeteredCreateBase(ProductPriceCreateBase): 1a
250 meter_id: UUID4 = Field(description="The ID of the meter associated to the price.") 1a
253class ProductPriceMeteredUnitCreate(ProductPriceMeteredCreateBase): 1a
254 """
255 Schema to create a metered price with a fixed unit price.
256 """
258 amount_type: Literal[ProductPriceAmountType.metered_unit] 1a
259 price_currency: PriceCurrency = "usd" 1a
260 unit_amount: Decimal = Field( 1a
261 gt=0,
262 max_digits=17,
263 decimal_places=12,
264 description="The price per unit in cents. Supports up to 12 decimal places.",
265 )
266 cap_amount: int | None = Field( 1a
267 default=None,
268 ge=0,
269 le=INT_MAX_VALUE,
270 description=(
271 "Optional maximum amount in cents that can be charged, "
272 "regardless of the number of units consumed."
273 ),
274 )
276 def get_model_class(self) -> builtins.type[ProductPriceMeteredUnitModel]: 1a
277 return ProductPriceMeteredUnitModel
280ProductPriceCreate = Annotated[ 1a
281 ProductPriceFixedCreate
282 | ProductPriceCustomCreate
283 | ProductPriceFreeCreate
284 | ProductPriceSeatBasedCreate
285 | ProductPriceMeteredUnitCreate,
286 Discriminator("amount_type"),
287]
290ProductPriceCreateList = Annotated[ 1a
291 list[ProductPriceCreate],
292 Field(min_length=1),
293 MergeJSONSchema(
294 {
295 "title": "ProductPriceCreateList",
296 "description": (
297 "List of prices for the product. "
298 "At most one static price (fixed, custom or free) is allowed. "
299 "Any number of metered prices can be added."
300 ),
301 }
302 ),
303]
306class ProductCreateBase(MetadataInputMixin, Schema): 1a
307 name: ProductName 1a
308 description: ProductDescription = None 1a
309 prices: ProductPriceCreateList = Field( 1a
310 ...,
311 description="List of available prices for this product. "
312 "It should contain at most one static price (fixed, custom or free), and "
313 "any number of metered prices. "
314 "Metered prices are not supported on one-time purchase products.",
315 )
316 medias: list[UUID4] | None = Field( 1a
317 default=None,
318 description=(
319 "List of file IDs. "
320 "Each one must be on the same organization as the product, "
321 "of type `product_media` and correctly uploaded."
322 ),
323 )
324 attached_custom_fields: AttachedCustomFieldListCreate = Field(default_factory=list) 1a
325 organization_id: OrganizationID | None = Field( 1a
326 default=None,
327 description=(
328 "The ID of the organization owning the product. "
329 "**Required unless you use an organization token.**"
330 ),
331 )
334class ProductCreateRecurring(TrialConfigurationInputMixin, ProductCreateBase): 1a
335 recurring_interval: SubscriptionRecurringInterval = Field( 1a
336 description=(
337 "The recurring interval of the product. "
338 "Note that the `day` and `week` values are for internal Polar staff use only."
339 ),
340 )
341 recurring_interval_count: int = Field( 1a
342 default=1,
343 ge=1,
344 le=999,
345 description=(
346 "Number of interval units of the subscription. "
347 "If this is set to 1 the charge will happen every interval (e.g. every month), "
348 "if set to 2 it will be every other month, and so on."
349 ),
350 )
353class ProductCreateOneTime(ProductCreateBase): 1a
354 recurring_interval: Literal[None] = Field( 1a
355 default=None, description="States that the product is a one-time purchase."
356 )
357 recurring_interval_count: Literal[None] = Field( 1a
358 default=None,
359 description="One-time products don't have a recurring interval count.",
360 )
363ProductCreate = Annotated[ 1a
364 ProductCreateRecurring | ProductCreateOneTime,
365 SetSchemaReference("ProductCreate"),
366]
369class ExistingProductPrice(Schema): 1a
370 """
371 A price that already exists for this product.
373 Useful when updating a product if you want to keep an existing price.
374 """
376 id: UUID4 1a
379ProductPriceUpdate = Annotated[ 1a
380 ExistingProductPrice | ProductPriceCreate, Field(union_mode="left_to_right")
381]
384class ProductUpdate(TrialConfigurationInputMixin, MetadataInputMixin, Schema): 1a
385 """
386 Schema to update a product.
387 """
389 name: ProductName | None = None 1a
390 description: ProductDescription = None 1a
391 recurring_interval: SubscriptionRecurringInterval | None = Field( 1a
392 default=None,
393 description=(
394 "The recurring interval of the product. "
395 "If `None`, the product is a one-time purchase. "
396 "**Can only be set on legacy recurring products. "
397 "Once set, it can't be changed.**"
398 ),
399 )
400 recurring_interval_count: int | None = Field( 1a
401 default=None,
402 ge=1,
403 le=999,
404 description=(
405 "Number of interval units of the subscription. "
406 "If this is set to 1 the charge will happen every interval (e.g. every month), "
407 "if set to 2 it will be every other month, and so on. "
408 "Once set, it can't be changed.**"
409 ),
410 )
411 is_archived: bool | None = Field( 1a
412 default=None,
413 description=(
414 "Whether the product is archived. "
415 "If `true`, the product won't be available for purchase anymore. "
416 "Existing customers will still have access to their benefits, "
417 "and subscriptions will continue normally."
418 ),
419 )
420 prices: list[ProductPriceUpdate] | None = Field( 1a
421 default=None,
422 description=(
423 "List of available prices for this product. "
424 "If you want to keep existing prices, include them in the list "
425 "as an `ExistingProductPrice` object."
426 ),
427 )
428 medias: list[UUID4] | None = Field( 1a
429 default=None,
430 description=(
431 "List of file IDs. "
432 "Each one must be on the same organization as the product, "
433 "of type `product_media` and correctly uploaded."
434 ),
435 )
436 attached_custom_fields: AttachedCustomFieldListCreate | None = None 1a
439class ProductBenefitsUpdate(Schema): 1a
440 """
441 Schema to update the benefits granted by a product.
442 """
444 benefits: list[BenefitID] = Field( 1a
445 description=(
446 "List of benefit IDs. "
447 "Each one must be on the same organization as the product."
448 )
449 )
452class ProductPriceBase(TimestampedSchema): 1a
453 id: UUID4 = Field(description="The ID of the price.") 1a
454 source: ProductPriceSource = Field( 1a
455 description=(
456 "The source of the price . "
457 "`catalog` is a predefined price, "
458 "while `ad_hoc` is a price created dynamically on a Checkout session."
459 )
460 )
461 amount_type: ProductPriceAmountType = Field( 1a
462 description="The type of amount, either fixed or custom."
463 )
464 is_archived: bool = Field( 1a
465 description="Whether the price is archived and no longer available."
466 )
467 product_id: UUID4 = Field(description="The ID of the product owning the price.") 1a
469 type: ProductPriceType = Field( 1a
470 validation_alias=AliasChoices("legacy_type", "type"),
471 deprecated=(
472 "This field is actually set from Product. "
473 "It's only kept for backward compatibility."
474 ),
475 )
476 recurring_interval: SubscriptionRecurringInterval | None = Field( 1a
477 validation_alias=AliasChoices(
478 "legacy_recurring_interval", "recurring_interval"
479 ),
480 deprecated=(
481 "This field is actually set from Product. "
482 "It's only kept for backward compatibility."
483 ),
484 )
487class ProductPriceFixedBase(ProductPriceBase): 1a
488 amount_type: Literal[ProductPriceAmountType.fixed] 1a
489 price_currency: str = Field(description="The currency.") 1a
490 price_amount: int = Field(description="The price in cents.") 1a
493class ProductPriceCustomBase(ProductPriceBase): 1a
494 amount_type: Literal[ProductPriceAmountType.custom] 1a
495 price_currency: str = Field(description="The currency.") 1a
496 minimum_amount: int | None = Field( 1a
497 description="The minimum amount the customer can pay."
498 )
499 maximum_amount: int | None = Field( 1a
500 description="The maximum amount the customer can pay."
501 )
502 preset_amount: int | None = Field( 1a
503 description="The initial amount shown to the customer."
504 )
507class ProductPriceFreeBase(ProductPriceBase): 1a
508 amount_type: Literal[ProductPriceAmountType.free] 1a
511class ProductPriceSeatBasedBase(ProductPriceBase): 1a
512 amount_type: Literal[ProductPriceAmountType.seat_based] 1a
513 price_currency: str = Field(description="The currency.") 1a
514 seat_tiers: ProductPriceSeatTiers = Field( 1a
515 description="Tiered pricing based on seat quantity"
516 )
518 @computed_field( 1a
519 description="Price per seat in cents from the first tier.",
520 deprecated=(
521 "Use `seat_tiers` instead. "
522 "The tiered pricing system supports volume-based pricing with multiple tiers. "
523 "This field returns only the first tier's price for backward compatibility."
524 ),
525 )
526 def price_per_seat(self) -> SkipJsonSchema[int]: 1a
527 """Return price_per_seat from first tier for backward compatibility."""
528 if not self.seat_tiers.tiers:
529 # This shouldn't happen due to validation, but protect against it
530 raise ValueError("seat_tiers must contain at least one tier")
531 return self.seat_tiers.tiers[0].price_per_seat
534class LegacyRecurringProductPriceMixin: 1a
535 @computed_field 1a
536 def legacy(self) -> Literal[True]: 1a
537 return True
540class LegacyRecurringProductPriceFixed( 1a
541 ProductPriceFixedBase, LegacyRecurringProductPriceMixin
542):
543 """
544 A recurring price for a product, i.e. a subscription.
546 **Deprecated**: The recurring interval should be set on the product itself.
547 """
549 type: Literal[ProductPriceType.recurring] = Field( 1a
550 description="The type of the price."
551 )
552 recurring_interval: SubscriptionRecurringInterval = Field( 1a
553 description="The recurring interval of the price."
554 )
557class LegacyRecurringProductPriceCustom( 1a
558 ProductPriceCustomBase, LegacyRecurringProductPriceMixin
559):
560 """
561 A pay-what-you-want recurring price for a product, i.e. a subscription.
563 **Deprecated**: The recurring interval should be set on the product itself.
564 """
566 type: Literal[ProductPriceType.recurring] = Field( 1a
567 description="The type of the price."
568 )
569 recurring_interval: SubscriptionRecurringInterval = Field( 1a
570 description="The recurring interval of the price."
571 )
574class LegacyRecurringProductPriceFree( 1a
575 ProductPriceFreeBase, LegacyRecurringProductPriceMixin
576):
577 """
578 A free recurring price for a product, i.e. a subscription.
580 **Deprecated**: The recurring interval should be set on the product itself.
581 """
583 type: Literal[ProductPriceType.recurring] = Field( 1a
584 description="The type of the price."
585 )
586 recurring_interval: SubscriptionRecurringInterval = Field( 1a
587 description="The recurring interval of the price."
588 )
591LegacyRecurringProductPrice = Annotated[ 1a
592 LegacyRecurringProductPriceFixed
593 | LegacyRecurringProductPriceCustom
594 | LegacyRecurringProductPriceFree,
595 Discriminator("amount_type"),
596 SetSchemaReference("LegacyRecurringProductPrice"),
597]
600class ProductPriceFixed(ProductPriceFixedBase): 1a
601 """
602 A fixed price for a product.
603 """
606class ProductPriceCustom(ProductPriceCustomBase): 1a
607 """
608 A pay-what-you-want price for a product.
609 """
612class ProductPriceFree(ProductPriceFreeBase): 1a
613 """
614 A free price for a product.
615 """
618class ProductPriceSeatBased(ProductPriceSeatBasedBase): 1a
619 """
620 A seat-based price for a product.
621 """
624class ProductPriceMeter(IDSchema): 1a
625 """
626 A meter associated to a metered price.
627 """
629 name: str = Field(description="The name of the meter.") 1a
632class ProductPriceMeteredUnit(ProductPriceBase): 1a
633 """
634 A metered, usage-based, price for a product, with a fixed unit price.
635 """
637 amount_type: Literal[ProductPriceAmountType.metered_unit] 1a
638 price_currency: str = Field(description="The currency.") 1a
639 unit_amount: Decimal = Field(description="The price per unit in cents.") 1a
640 cap_amount: int | None = Field( 1a
641 description=(
642 "The maximum amount in cents that can be charged, "
643 "regardless of the number of units consumed."
644 )
645 )
646 meter_id: UUID4 = Field(description="The ID of the meter associated to the price.") 1a
647 meter: ProductPriceMeter = Field(description="The meter associated to the price.") 1a
650NewProductPrice = Annotated[ 1a
651 ProductPriceFixed
652 | ProductPriceCustom
653 | ProductPriceFree
654 | ProductPriceSeatBased
655 | ProductPriceMeteredUnit,
656 Discriminator("amount_type"),
657 SetSchemaReference("ProductPrice"),
658]
661def _get_discriminator_value(v: Any) -> Literal["legacy", "new"]: 1a
662 if isinstance(v, dict):
663 return "legacy" if "legacy" in v else "new"
664 type = getattr(v, "type", None)
665 return "legacy" if type is not None else "new"
668ProductPrice = Annotated[ 1a
669 Annotated[LegacyRecurringProductPrice, Tag("legacy")]
670 | Annotated[NewProductPrice, Tag("new")],
671 Discriminator(_get_discriminator_value),
672]
675class ProductBase(TrialConfigurationOutputMixin, TimestampedSchema, IDSchema): 1a
676 name: str = Field(description="The name of the product.") 1a
677 description: str | None = Field(description="The description of the product.") 1a
678 recurring_interval: SubscriptionRecurringInterval | None = Field( 1a
679 description=(
680 "The recurring interval of the product. "
681 "If `None`, the product is a one-time purchase."
682 )
683 )
684 recurring_interval_count: int | None = Field( 1a
685 description=(
686 "Number of interval units of the subscription. "
687 "If this is set to 1 the charge will happen every interval (e.g. every month), "
688 "if set to 2 it will be every other month, and so on. "
689 "None for one-time products."
690 )
691 )
692 is_recurring: bool = Field(description="Whether the product is a subscription.") 1a
693 is_archived: bool = Field( 1a
694 description="Whether the product is archived and no longer available."
695 )
696 organization_id: UUID4 = Field( 1a
697 description="The ID of the organization owning the product."
698 )
701ProductPriceList = Annotated[ 1a
702 list[ProductPrice],
703 Field(
704 description="List of prices for this product.",
705 ),
706]
707BenefitList = Annotated[ 1a
708 list[Benefit],
709 Field(
710 description="List of benefits granted by the product.",
711 ),
712]
713ProductMediaList = Annotated[ 1a
714 list[ProductMediaFileRead],
715 Field(
716 description="List of medias associated to the product.",
717 ),
718]
721class Product(MetadataOutputMixin, ProductBase): 1a
722 """
723 A product.
724 """
726 prices: ProductPriceList 1a
727 benefits: BenefitList 1a
728 medias: ProductMediaList 1a
729 attached_custom_fields: list[AttachedCustomField] = Field( 1a
730 description="List of custom fields attached to the product."
731 )
734BenefitPublicList = Annotated[ 1a
735 list[BenefitPublic],
736 Field(
737 title="BenefitPublic",
738 description="List of benefits granted by the product.",
739 ),
740]