Coverage for polar/customer_portal/schemas/order.py: 98%

1from datetime import datetime 

2 

3from pydantic import UUID4, AliasChoices, AliasPath, Field, model_validator 

4from pydantic.json_schema import SkipJsonSchema 

5 

6from polar.enums import PaymentProcessor 

7from polar.kit.schemas import Schema 

8from polar.order.schemas import OrderBase, OrderItemSchema, OrderUpdateBase 

9from polar.product.schemas import ( 

10 BenefitPublicList, 

11 ProductBase, 

12 ProductMediaList, 

13 ProductPrice, 

14 ProductPriceList, 

15) 

16from polar.subscription.schemas import SubscriptionBase 

17 

18from .organization import CustomerOrganization 

19 

20 

21class CustomerOrderProduct(ProductBase): 

22 prices: ProductPriceList 

23 benefits: BenefitPublicList 

24 medias: ProductMediaList 

25 organization: CustomerOrganization 

26 

27 

28class CustomerOrderSubscription(SubscriptionBase): ... 28 ↛ 31line 28 didn't jump to line 31 because

29 

30 

31class CustomerOrder(OrderBase): 

32 user_id: UUID4 = Field( 

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 

42 product_price: SkipJsonSchema[ProductPrice | None] = Field( 

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 

52 items: list[OrderItemSchema] = Field(description="Line items composing the order.") 

53 description: str = Field( 

54 description="A summary description of the order.", examples=["Pro Plan"] 

55 ) 

56 next_payment_attempt_at: datetime | None = Field( 

57 None, description="When the next payment retry is scheduled" 

58 ) 

59 

60 

61class CustomerOrderInvoice(Schema): 

62 """Order's invoice data.""" 

63 

64 url: str = Field(..., description="The URL to the invoice.") 

65 

66 

67class CustomerOrderUpdate(OrderUpdateBase): 

68 """Schema to update an order.""" 

69 

70 

71class CustomerOrderPaymentStatus(Schema): 

72 """Payment status for an order.""" 

73 

74 status: str = Field(..., description="Current payment status.") 

75 error: str | None | None = Field( 

76 None, description="Error message if payment failed." 

77 ) 

78 

79 

80class CustomerOrderConfirmPayment(Schema): 

81 """Schema to confirm a retry payment using either a saved payment method or a new confirmation token.""" 

82 

83 confirmation_token_id: str | None = Field( 

84 None, description="ID of the Stripe confirmation token for new payment methods." 

85 ) 

86 payment_method_id: UUID4 | None = Field( 

87 None, description="ID of an existing saved payment method." 

88 ) 

89 payment_processor: PaymentProcessor = Field( 

90 PaymentProcessor.stripe, description="Payment processor used." 

91 ) 

92 

93 @model_validator(mode="after") 

94 def validate_payment_method(self) -> "CustomerOrderConfirmPayment": 

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 

108 

109 

110class CustomerOrderPaymentConfirmation(Schema): 

111 """Response after confirming a retry payment.""" 

112 

113 status: str = Field(..., description="Payment status after confirmation.") 

114 client_secret: str | None = Field( 

115 None, description="Client secret for handling additional actions." 

116 ) 

117 error: str | None = Field(None, description="Error message if confirmation failed.")