Coverage for polar/license_key/schemas.py: 71%
112 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 16:17 +0000
1from datetime import datetime 1a
2from typing import Annotated, Literal, Self 1a
4from dateutil.relativedelta import relativedelta 1a
5from pydantic import UUID4, AliasChoices, AliasPath, Field 1a
6from pydantic.json_schema import SkipJsonSchema 1a
8from polar.benefit.schemas import BenefitID 1a
9from polar.benefit.strategies.license_keys.properties import ( 1a
10 BenefitLicenseKeyActivationProperties,
11 BenefitLicenseKeyExpirationProperties,
12)
13from polar.customer.schemas.customer import CustomerBase 1a
14from polar.exceptions import NotPermitted, ResourceNotFound, Unauthorized 1a
15from polar.kit.metadata import ( 1a
16 MAXIMUM_KEYS,
17 METADATA_DESCRIPTION,
18 MetadataKey,
19 MetadataValue,
20)
21from polar.kit.schemas import IDSchema, Schema, TimestampedSchema 1a
22from polar.kit.utils import generate_uuid, utc_now 1a
23from polar.models.license_key import LicenseKeyStatus 1a
25###############################################################################
26# RESPONSES
27###############################################################################
29NotFoundResponse = { 1a
30 "description": "License key not found.",
31 "model": ResourceNotFound.schema(),
32}
34UnauthorizedResponse = { 1a
35 "description": "Not authorized to manage license key.",
36 "model": Unauthorized.schema(),
37}
40ActivationNotPermitted = { 1a
41 "description": "License key activation not supported or limit reached. Use /validate endpoint for licenses without activations.",
42 "model": NotPermitted.schema(),
43}
46###############################################################################
47# RESPONSES
48###############################################################################
50ConditionsField = Annotated[ 1a
51 dict[MetadataKey, MetadataValue],
52 Field(
53 default_factory=dict,
54 max_length=MAXIMUM_KEYS,
55 description=METADATA_DESCRIPTION.format(
56 heading=(
57 "Key-value object allowing you to set conditions "
58 "that must match when validating the license key."
59 )
60 ),
61 ),
62]
64MetaField = Annotated[ 1a
65 dict[MetadataKey, MetadataValue],
66 Field(
67 default_factory=dict,
68 max_length=MAXIMUM_KEYS,
69 description=METADATA_DESCRIPTION.format(
70 heading=(
71 "Key-value object allowing you to store additional information "
72 "about the activation"
73 )
74 ),
75 ),
76]
79class LicenseKeyValidate(Schema): 1a
80 key: str 1a
81 organization_id: UUID4 1a
82 activation_id: UUID4 | None = None 1a
83 benefit_id: BenefitID | None = None 1a
84 customer_id: UUID4 | None = None 1a
85 increment_usage: int | None = None 1a
86 conditions: ConditionsField 1a
89class LicenseKeyActivate(Schema): 1a
90 key: str 1a
91 organization_id: UUID4 1a
92 label: str 1a
93 conditions: ConditionsField 1a
94 meta: MetaField 1a
97class LicenseKeyDeactivate(Schema): 1a
98 key: str 1a
99 organization_id: UUID4 1a
100 activation_id: UUID4 1a
103class LicenseKeyCustomer(CustomerBase): ... 1a
106class LicenseKeyUser(Schema): 1a
107 id: UUID4 = Field( 1a
108 validation_alias=AliasChoices(
109 # Validate from ORM model
110 "legacy_user_id",
111 # Validate from stored webhook payload
112 "id",
113 )
114 )
115 email: str 1a
116 public_name: str = Field( 1a
117 validation_alias=AliasChoices(
118 # Validate from ORM model
119 "legacy_user_public_name",
120 # Validate from stored webhook payload
121 "public_name",
122 )
123 )
124 avatar_url: str | None = Field(None) 1a
127class LicenseKeyRead(TimestampedSchema, IDSchema): 1a
128 organization_id: UUID4 1a
129 user_id: SkipJsonSchema[UUID4] = Field( 1a
130 validation_alias=AliasChoices(
131 # Validate from stored webhook payload
132 "user_id",
133 # Validate from ORM model
134 AliasPath("customer", "legacy_user_id"),
135 ),
136 deprecated="Use `customer_id`.",
137 )
138 customer_id: UUID4 1a
139 user: SkipJsonSchema[LicenseKeyUser] = Field( 1a
140 validation_alias=AliasChoices(
141 # Validate from stored webhook payload
142 "user",
143 # Validate from ORM model
144 "customer",
145 ),
146 deprecated="Use `customer`.",
147 )
148 customer: LicenseKeyCustomer 1a
149 benefit_id: BenefitID 1a
150 key: str 1a
151 display_key: str 1a
152 status: LicenseKeyStatus 1a
153 limit_activations: int | None 1a
154 usage: int 1a
155 limit_usage: int | None 1a
156 validations: int 1a
157 last_validated_at: datetime | None 1a
158 expires_at: datetime | None 1a
161class LicenseKeyActivationBase(Schema): 1a
162 id: UUID4 1a
163 license_key_id: UUID4 1a
164 label: str 1a
165 meta: dict[str, str | int | float | bool] 1a
166 created_at: datetime 1a
167 modified_at: datetime | None 1a
170class LicenseKeyWithActivations(LicenseKeyRead): 1a
171 activations: list[LicenseKeyActivationBase] 1a
174class ValidatedLicenseKey(LicenseKeyRead): 1a
175 activation: LicenseKeyActivationBase | None = None 1a
178class LicenseKeyActivationRead(LicenseKeyActivationBase): 1a
179 license_key: LicenseKeyRead 1a
182class LicenseKeyUpdate(Schema): 1a
183 status: LicenseKeyStatus | None = None 1a
184 usage: int = 0 1a
185 limit_activations: int | None = Field(gt=0, le=1000, default=None) 1a
186 limit_usage: int | None = Field(gt=0, default=None) 1a
187 expires_at: datetime | None = None 1a
190class LicenseKeyCreate(LicenseKeyUpdate): 1a
191 organization_id: UUID4 1a
192 customer_id: UUID4 1a
193 benefit_id: BenefitID 1a
194 key: str 1a
196 @classmethod 1a
197 def generate_key(cls, prefix: str | None = None) -> str: 1a
198 key = str(generate_uuid()).upper()
199 if prefix is None:
200 return key
202 prefix = prefix.strip().upper()
203 return f"{prefix}-{key}"
205 @classmethod 1a
206 def generate_expiration_dt( 1a
207 cls, ttl: int, timeframe: Literal["year", "month", "day"]
208 ) -> datetime:
209 now = utc_now()
210 match timeframe:
211 case "year":
212 return now + relativedelta(years=ttl)
213 case "month":
214 return now + relativedelta(months=ttl)
215 case _:
216 return now + relativedelta(days=ttl)
218 @classmethod 1a
219 def build( 1a
220 cls,
221 organization_id: UUID4,
222 customer_id: UUID4,
223 benefit_id: UUID4,
224 prefix: str | None = None,
225 status: LicenseKeyStatus = LicenseKeyStatus.granted,
226 limit_usage: int | None = None,
227 activations: BenefitLicenseKeyActivationProperties | None = None,
228 expires: BenefitLicenseKeyExpirationProperties | None = None,
229 ) -> Self:
230 expires_at = None
231 if expires:
232 ttl = expires.get("ttl", None)
233 timeframe = expires.get("timeframe", None)
234 if ttl and timeframe:
235 expires_at = cls.generate_expiration_dt(ttl, timeframe)
237 limit_activations = None
238 if activations:
239 limit_activations = activations.get("limit", None)
241 key = cls.generate_key(prefix=prefix)
242 return cls(
243 organization_id=organization_id,
244 customer_id=customer_id,
245 benefit_id=benefit_id,
246 key=key,
247 status=status,
248 limit_activations=limit_activations,
249 limit_usage=limit_usage,
250 expires_at=expires_at,
251 )