Coverage for polar/customer_portal/schemas/order.py: 82%
44 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 16:17 +0000
1from datetime import datetime 1a
3from pydantic import UUID4, AliasChoices, AliasPath, Field, model_validator 1a
4from pydantic.json_schema import SkipJsonSchema 1a
6from polar.enums import PaymentProcessor 1a
7from polar.kit.schemas import Schema 1a
8from polar.order.schemas import OrderBase, OrderItemSchema, OrderUpdateBase 1a
9from polar.product.schemas import ( 1a
10 BenefitPublicList,
11 ProductBase,
12 ProductMediaList,
13 ProductPrice,
14 ProductPriceList,
15)
16from polar.subscription.schemas import SubscriptionBase 1a
18from .organization import CustomerOrganization 1a
21class CustomerOrderProduct(ProductBase): 1a
22 prices: ProductPriceList 1a
23 benefits: BenefitPublicList 1a
24 medias: ProductMediaList 1a
25 organization: CustomerOrganization 1a
28class CustomerOrderSubscription(SubscriptionBase): ... 1a
31class CustomerOrder(OrderBase): 1a
32 user_id: UUID4 = Field( 1a
33 validation_alias=AliasChoices(
34 # Validate from stored webhook payload
35 "user_id",
36 # Validate from ORM model
37 AliasPath("customer", "legacy_user_id"),
38 ),
39 deprecated="Use `customer_id`.",
40 )
41 product: CustomerOrderProduct | None 1a
42 product_price: SkipJsonSchema[ProductPrice | None] = Field( 1a
43 deprecated="Use `items` instead.",
44 validation_alias=AliasChoices(
45 # Validate from stored webhook payload
46 "product_price",
47 # Validate from ORM model
48 "legacy_product_price",
49 ),
50 )
51 subscription: CustomerOrderSubscription | None 1a
52 items: list[OrderItemSchema] = Field(description="Line items composing the order.") 1a
53 description: str = Field( 1a
54 description="A summary description of the order.", examples=["Pro Plan"]
55 )
56 next_payment_attempt_at: datetime | None = Field( 1a
57 None, description="When the next payment retry is scheduled"
58 )
61class CustomerOrderInvoice(Schema): 1a
62 """Order's invoice data."""
64 url: str = Field(..., description="The URL to the invoice.") 1a
67class CustomerOrderUpdate(OrderUpdateBase): 1a
68 """Schema to update an order."""
71class CustomerOrderPaymentStatus(Schema): 1a
72 """Payment status for an order."""
74 status: str = Field(..., description="Current payment status.") 1a
75 error: str | None | None = Field( 1a
76 None, description="Error message if payment failed."
77 )
80class CustomerOrderConfirmPayment(Schema): 1a
81 """Schema to confirm a retry payment using either a saved payment method or a new confirmation token."""
83 confirmation_token_id: str | None = Field( 1a
84 None, description="ID of the Stripe confirmation token for new payment methods."
85 )
86 payment_method_id: UUID4 | None = Field( 1a
87 None, description="ID of an existing saved payment method."
88 )
89 payment_processor: PaymentProcessor = Field( 1a
90 PaymentProcessor.stripe, description="Payment processor used."
91 )
93 @model_validator(mode="after") 1a
94 def validate_payment_method(self) -> "CustomerOrderConfirmPayment": 1a
95 """Ensure exactly one of confirmation_token_id or payment_method_id is provided."""
96 if self.confirmation_token_id is None and self.payment_method_id is None:
97 raise ValueError(
98 "Either confirmation_token_id or payment_method_id must be provided"
99 )
100 if (
101 self.confirmation_token_id is not None
102 and self.payment_method_id is not None
103 ):
104 raise ValueError(
105 "Only one of confirmation_token_id or payment_method_id can be provided"
106 )
107 return self
110class CustomerOrderPaymentConfirmation(Schema): 1a
111 """Response after confirming a retry payment."""
113 status: str = Field(..., description="Payment status after confirmation.") 1a
114 client_secret: str | None = Field( 1a
115 None, description="Client secret for handling additional actions."
116 )
117 error: str | None = Field(None, description="Error message if confirmation failed.") 1a