Coverage for polar/customer/schemas/state.py: 100%
49 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 17:15 +0000
1from datetime import datetime 1a
2from typing import Literal 1a
4from pydantic import UUID4, AliasChoices, Field 1a
5from pydantic.aliases import AliasPath 1a
6from pydantic.json_schema import SkipJsonSchema 1a
8from polar.benefit.strategies import BenefitGrantProperties 1a
9from polar.custom_field.data import CustomFieldDataOutputMixin 1a
10from polar.enums import SubscriptionRecurringInterval 1a
11from polar.kit.metadata import ( 1a
12 MetadataOutputMixin,
13 MetadataOutputType,
14)
15from polar.kit.schemas import ( 1a
16 BENEFIT_GRANT_ID_EXAMPLE,
17 BENEFIT_ID_EXAMPLE,
18 METER_ID_EXAMPLE,
19 PRICE_ID_EXAMPLE,
20 PRODUCT_ID_EXAMPLE,
21 SUBSCRIPTION_ID_EXAMPLE,
22 IDSchema,
23 TimestampedSchema,
24)
25from polar.models.benefit import BenefitType 1a
26from polar.models.subscription import SubscriptionStatus 1a
27from polar.subscription.schemas import SubscriptionMeterBase 1a
29from .customer import CustomerBase 1a
32class CustomerStateSubscriptionMeter(SubscriptionMeterBase): 1a
33 """Current consumption and spending for a subscription meter."""
36class CustomerStateSubscription( 1a
37 MetadataOutputMixin, CustomFieldDataOutputMixin, TimestampedSchema, IDSchema
38):
39 """An active customer subscription."""
41 id: UUID4 = Field( 1a
42 description="The ID of the subscription.", examples=[SUBSCRIPTION_ID_EXAMPLE]
43 )
44 status: Literal[SubscriptionStatus.active, SubscriptionStatus.trialing] = Field( 1a
45 examples=["active", "trialing"]
46 )
47 amount: int = Field(description="The amount of the subscription.", examples=[1000]) 1a
48 currency: str = Field( 1a
49 description="The currency of the subscription.", examples=["usd"]
50 )
51 recurring_interval: SubscriptionRecurringInterval = Field( 1a
52 description="The interval at which the subscription recurs."
53 )
54 current_period_start: datetime = Field( 1a
55 description="The start timestamp of the current billing period.",
56 examples=["2025-02-03T13:37:00Z"],
57 )
58 current_period_end: datetime | None = Field( 1a
59 description="The end timestamp of the current billing period.",
60 examples=["2025-03-03T13:37:00Z"],
61 )
62 trial_start: datetime | None = Field( 1a
63 description="The start timestamp of the trial period, if any.",
64 examples=["2025-02-03T13:37:00Z"],
65 )
66 trial_end: datetime | None = Field( 1a
67 description="The end timestamp of the trial period, if any.",
68 examples=["2025-03-03T13:37:00Z"],
69 )
70 cancel_at_period_end: bool = Field( 1a
71 description=(
72 "Whether the subscription will be canceled "
73 "at the end of the current period."
74 ),
75 examples=[False],
76 )
77 canceled_at: datetime | None = Field( 1a
78 description=(
79 "The timestamp when the subscription was canceled. "
80 "The subscription might still be active if `cancel_at_period_end` is `true`."
81 ),
82 examples=[None],
83 )
84 started_at: datetime | None = Field( 1a
85 description="The timestamp when the subscription started.",
86 examples=["2025-01-03T13:37:00Z"],
87 )
88 ends_at: datetime | None = Field( 1a
89 description="The timestamp when the subscription will end.",
90 examples=[None],
91 )
93 product_id: UUID4 = Field( 1a
94 description="The ID of the subscribed product.", examples=[PRODUCT_ID_EXAMPLE]
95 )
96 discount_id: UUID4 | None = Field( 1a
97 description="The ID of the applied discount, if any.", examples=[None]
98 )
100 price_id: SkipJsonSchema[UUID4] = Field( 1a
101 deprecated=True,
102 examples=[PRICE_ID_EXAMPLE],
103 validation_alias=AliasChoices(
104 # Validate from stored webhook payload
105 "price_id",
106 # Validate from ORM model
107 AliasPath("prices", 0, "id"),
108 ),
109 )
110 meters: list[CustomerStateSubscriptionMeter] = Field( 1a
111 description="List of meters associated with the subscription."
112 )
115class CustomerStateBenefitGrant(TimestampedSchema, IDSchema): 1a
116 """An active benefit grant for a customer."""
118 id: UUID4 = Field( 1a
119 description="The ID of the grant.", examples=[BENEFIT_GRANT_ID_EXAMPLE]
120 )
121 granted_at: datetime = Field( 1a
122 description="The timestamp when the benefit was granted.",
123 examples=["2025-01-03T13:37:00Z"],
124 )
125 benefit_id: UUID4 = Field( 1a
126 description="The ID of the benefit concerned by this grant.",
127 examples=[BENEFIT_ID_EXAMPLE],
128 )
129 benefit_type: BenefitType = Field( 1a
130 description="The type of the benefit concerned by this grant.",
131 validation_alias=AliasChoices(
132 # Validate from stored webhook payload
133 "benefit_type",
134 # Validate from ORM model
135 AliasPath("benefit", "type"),
136 ),
137 examples=[BenefitType.custom],
138 )
139 benefit_metadata: MetadataOutputType = Field( 1a
140 description="The metadata of the benefit concerned by this grant.",
141 examples=[{"key": "value"}],
142 validation_alias=AliasChoices(
143 # Validate from stored webhook payload
144 "benefit_metadata",
145 # Validate from ORM model
146 AliasPath("benefit", "user_metadata"),
147 ),
148 )
149 properties: BenefitGrantProperties 1a
152class CustomerStateMeter(TimestampedSchema, IDSchema): 1a
153 """An active meter for a customer, with latest consumed and credited units."""
155 meter_id: UUID4 = Field( 1a
156 description="The ID of the meter.", examples=[METER_ID_EXAMPLE]
157 )
158 consumed_units: float = Field( 1a
159 description="The number of consumed units.", examples=[25.0]
160 )
161 credited_units: int = Field( 1a
162 description="The number of credited units.", examples=[100]
163 )
164 balance: float = Field( 1a
165 description=(
166 "The balance of the meter, "
167 "i.e. the difference between credited and consumed units."
168 ),
169 examples=[75.0],
170 )
173class CustomerState(CustomerBase): 1a
174 """
175 A customer along with additional state information:
177 * Active subscriptions
178 * Granted benefits
179 * Active meters
180 """
182 active_subscriptions: list[CustomerStateSubscription] = Field( 1a
183 description="The customer's active subscriptions."
184 )
185 granted_benefits: list[CustomerStateBenefitGrant] = Field( 1a
186 description="The customer's active benefit grants."
187 )
188 active_meters: list[CustomerStateMeter] = Field( 1a
189 description="The customer's active meters.",
190 )