Coverage for polar/organization/schemas.py: 96%
161 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 12:42 +0000
1from datetime import datetime
2from enum import StrEnum
3from typing import Annotated, Any, Literal
5from pydantic import (
6 UUID4,
7 AfterValidator,
8 BeforeValidator,
9 EmailStr,
10 Field,
11 StringConstraints,
12 model_validator,
13)
14from pydantic.json_schema import SkipJsonSchema
15from pydantic.networks import HttpUrl
17from polar.config import settings
18from polar.enums import SubscriptionProrationBehavior
19from polar.kit.email import EmailStrDNS
20from polar.kit.schemas import (
21 ORGANIZATION_ID_EXAMPLE,
22 EmptyStrToNoneValidator,
23 HttpUrlToStr,
24 IDSchema,
25 MergeJSONSchema,
26 Schema,
27 SelectorWidget,
28 SlugValidator,
29 TimestampedSchema,
30)
31from polar.models.organization import (
32 OrganizationCustomerEmailSettings,
33 OrganizationNotificationSettings,
34 OrganizationStatus,
35 OrganizationSubscriptionSettings,
36)
37from polar.models.organization_review import OrganizationReview
39OrganizationID = Annotated[
40 UUID4,
41 MergeJSONSchema({"description": "The organization ID."}),
42 SelectorWidget("/v1/organizations", "Organization", "name"),
43 Field(examples=[ORGANIZATION_ID_EXAMPLE]),
44]
46NameInput = Annotated[str, StringConstraints(min_length=3)]
49def validate_reserved_keywords(value: str) -> str:
50 if value in settings.ORGANIZATION_SLUG_RESERVED_KEYWORDS: 50 ↛ 51line 50 didn't jump to line 51 because the condition on line 50 was never true
51 raise ValueError("This slug is reserved.")
52 return value
55SlugInput = Annotated[
56 str,
57 StringConstraints(to_lower=True, min_length=3),
58 SlugValidator,
59 AfterValidator(validate_reserved_keywords),
60]
63class OrganizationFeatureSettings(Schema):
64 issue_funding_enabled: bool = Field(
65 False, description="If this organization has issue funding enabled"
66 )
67 seat_based_pricing_enabled: bool = Field(
68 False, description="If this organization has seat-based pricing enabled"
69 )
70 revops_enabled: bool = Field(
71 False, description="If this organization has RevOps enabled"
72 )
73 wallets_enabled: bool = Field(
74 False, description="If this organization has Wallets enabled"
75 )
76 member_model_enabled: bool = Field(
77 False, description="If this organization has the Member model enabled"
78 )
81class OrganizationSubscribePromoteSettings(Schema):
82 promote: bool = Field(True, description="Promote email subscription (free)")
83 show_count: bool = Field(True, description="Show subscription count publicly")
84 count_free: bool = Field(
85 True, description="Include free subscribers in total count"
86 )
89class OrganizationDetails(Schema):
90 about: str = Field(
91 ..., description="Brief information about you and your business."
92 )
93 product_description: str = Field(
94 ..., description="Description of digital products being sold."
95 )
96 intended_use: str = Field(
97 ..., description="How the organization will integrate and use Polar."
98 )
99 customer_acquisition: list[str] = Field(
100 ..., description="Main customer acquisition channels."
101 )
102 future_annual_revenue: int = Field(
103 ..., ge=0, description="Estimated revenue in the next 12 months"
104 )
105 switching: bool = Field(True, description="Switching from another platform?")
106 switching_from: (
107 Literal["paddle", "lemon_squeezy", "gumroad", "stripe", "other"] | None
108 ) = Field(None, description="Which platform the organization is migrating from.")
109 previous_annual_revenue: int = Field(
110 0, ge=0, description="Revenue from last year if applicable."
111 )
114class OrganizationSocialPlatforms(StrEnum):
115 x = "x"
116 github = "github"
117 facebook = "facebook"
118 instagram = "instagram"
119 youtube = "youtube"
120 tiktok = "tiktok"
121 linkedin = "linkedin"
122 other = "other"
125PLATFORM_DOMAINS = {
126 "x": ["twitter.com", "x.com"],
127 "github": ["github.com"],
128 "facebook": ["facebook.com", "fb.com"],
129 "instagram": ["instagram.com"],
130 "youtube": ["youtube.com", "youtu.be"],
131 "tiktok": ["tiktok.com"],
132 "linkedin": ["linkedin.com"],
133}
136class OrganizationSocialLink(Schema):
137 platform: OrganizationSocialPlatforms = Field(
138 ..., description="The social platform of the URL"
139 )
140 url: HttpUrlToStr = Field(..., description="The URL to the organization profile")
142 @model_validator(mode="before")
143 @classmethod
144 def validate_url(cls, data: dict[str, Any]) -> dict[str, Any]:
145 platform = data.get("platform")
146 url = data.get("url", "").lower()
148 if not (platform and url):
149 return data
151 if platform == "other":
152 return data
154 valid_domains = PLATFORM_DOMAINS[platform]
155 if not any(domain in url for domain in valid_domains): 155 ↛ 160line 155 didn't jump to line 160 because the condition on line 155 was always true
156 raise ValueError(
157 f"Invalid URL for {platform}. Must be from: {', '.join(valid_domains)}"
158 )
160 return data
163# Deprecated
164class OrganizationProfileSettings(Schema):
165 enabled: bool | None = Field(
166 None, description="If this organization has a profile enabled"
167 )
168 description: Annotated[
169 str | None,
170 Field(max_length=160, description="A description of the organization"),
171 EmptyStrToNoneValidator,
172 ] = None
173 featured_projects: list[UUID4] | None = Field(
174 None, description="A list of featured projects"
175 )
176 featured_organizations: list[UUID4] | None = Field(
177 None, description="A list of featured organizations"
178 )
179 links: list[HttpUrl] | None = Field(
180 None, description="A list of links associated with the organization"
181 )
182 subscribe: OrganizationSubscribePromoteSettings | None = Field(
183 OrganizationSubscribePromoteSettings(
184 promote=True,
185 show_count=True,
186 count_free=True,
187 ),
188 description="Subscription promotion settings",
189 )
190 accent_color: str | None = Field(
191 None, description="Accent color for the organization"
192 )
195class OrganizationBase(IDSchema, TimestampedSchema):
196 name: str = Field(
197 description="Organization name shown in checkout, customer portal, emails etc.",
198 )
199 slug: str = Field(
200 description="Unique organization slug in checkout, customer portal and credit card statements.",
201 )
202 avatar_url: str | None = Field(
203 description="Avatar URL shown in checkout, customer portal, emails etc."
204 )
205 proration_behavior: SubscriptionProrationBehavior = Field(
206 description="Proration behavior applied when customer updates their subscription from the portal.",
207 )
208 allow_customer_updates: bool = Field(
209 description="Whether customers can update their subscriptions from the customer portal.",
210 )
212 # Deprecated attributes
213 bio: SkipJsonSchema[str | None] = Field(..., deprecated="")
214 company: SkipJsonSchema[str | None] = Field(
215 ...,
216 deprecated="Legacy attribute no longer in use.",
217 )
218 blog: SkipJsonSchema[str | None] = Field(
219 ...,
220 deprecated="Legacy attribute no longer in use. See `socials` instead.",
221 )
222 location: SkipJsonSchema[str | None] = Field(
223 ...,
224 deprecated="Legacy attribute no longer in use.",
225 )
226 twitter_username: SkipJsonSchema[str | None] = Field(
227 ...,
228 deprecated="Legacy attribute no longer in use. See `socials` instead.",
229 )
231 pledge_minimum_amount: SkipJsonSchema[int] = Field(0, deprecated=True)
232 pledge_badge_show_amount: SkipJsonSchema[bool] = Field(False, deprecated=True)
233 default_upfront_split_to_contributors: SkipJsonSchema[int | None] = Field(
234 None, deprecated=True
235 )
236 profile_settings: SkipJsonSchema[OrganizationProfileSettings | None] = Field(
237 None, deprecated=True
238 )
241class LegacyOrganizationStatus(StrEnum):
242 """
243 Legacy organization status values kept for backward compatibility in schemas
244 using OrganizationPublicBase.
245 """
247 CREATED = "created"
248 ONBOARDING_STARTED = "onboarding_started"
249 UNDER_REVIEW = "under_review"
250 DENIED = "denied"
251 ACTIVE = "active"
253 @classmethod
254 def from_status(cls, status: OrganizationStatus) -> "LegacyOrganizationStatus":
255 mapping = {
256 OrganizationStatus.CREATED: LegacyOrganizationStatus.CREATED,
257 OrganizationStatus.ONBOARDING_STARTED: (
258 LegacyOrganizationStatus.ONBOARDING_STARTED
259 ),
260 OrganizationStatus.INITIAL_REVIEW: LegacyOrganizationStatus.UNDER_REVIEW,
261 OrganizationStatus.ONGOING_REVIEW: LegacyOrganizationStatus.UNDER_REVIEW,
262 OrganizationStatus.DENIED: LegacyOrganizationStatus.DENIED,
263 OrganizationStatus.ACTIVE: LegacyOrganizationStatus.ACTIVE,
264 }
265 try:
266 return mapping[status]
267 except KeyError as e:
268 raise ValueError("Unknown OrganizationStatus") from e
271class OrganizationPublicBase(OrganizationBase):
272 # Attributes that we used to have publicly, but now want to hide from
273 # the public schema.
274 # Keep it for now for backward compatibility in the SDK
275 email: SkipJsonSchema[str | None]
276 website: SkipJsonSchema[str | None]
277 socials: SkipJsonSchema[list[OrganizationSocialLink]]
278 status: Annotated[
279 SkipJsonSchema[LegacyOrganizationStatus],
280 BeforeValidator(LegacyOrganizationStatus.from_status),
281 ]
282 details_submitted_at: SkipJsonSchema[datetime | None]
284 feature_settings: SkipJsonSchema[OrganizationFeatureSettings | None]
285 subscription_settings: SkipJsonSchema[OrganizationSubscriptionSettings]
286 notification_settings: SkipJsonSchema[OrganizationNotificationSettings]
287 customer_email_settings: SkipJsonSchema[OrganizationCustomerEmailSettings]
290class Organization(OrganizationBase):
291 email: str | None = Field(description="Public support email.")
292 website: str | None = Field(description="Official website of the organization.")
293 socials: list[OrganizationSocialLink] = Field(
294 description="Links to social profiles.",
295 )
296 status: OrganizationStatus = Field(description="Current organization status")
297 details_submitted_at: datetime | None = Field(
298 description="When the business details were submitted.",
299 )
301 feature_settings: OrganizationFeatureSettings | None = Field(
302 description="Organization feature settings",
303 )
304 subscription_settings: OrganizationSubscriptionSettings = Field(
305 description="Settings related to subscriptions management",
306 )
307 notification_settings: OrganizationNotificationSettings = Field(
308 description="Settings related to notifications",
309 )
310 customer_email_settings: OrganizationCustomerEmailSettings = Field(
311 description="Settings related to customer emails",
312 )
315class OrganizationCreate(Schema):
316 name: NameInput
317 slug: SlugInput
318 avatar_url: HttpUrlToStr | None = None
319 email: EmailStr | None = Field(None, description="Public support email.")
320 website: HttpUrlToStr | None = Field(
321 None, description="Official website of the organization."
322 )
323 socials: list[OrganizationSocialLink] | None = Field(
324 None,
325 description="Link to social profiles.",
326 )
327 details: OrganizationDetails | None = Field(
328 None,
329 description="Additional, private, business details Polar needs about active organizations for compliance (KYC).",
330 )
331 feature_settings: OrganizationFeatureSettings | None = None
332 subscription_settings: OrganizationSubscriptionSettings | None = None
333 notification_settings: OrganizationNotificationSettings | None = None
334 customer_email_settings: OrganizationCustomerEmailSettings | None = None
337class OrganizationUpdate(Schema):
338 name: NameInput | None = None
339 avatar_url: HttpUrlToStr | None = None
341 email: EmailStrDNS | None = Field(None, description="Public support email.")
342 website: HttpUrlToStr | None = Field(
343 None, description="Official website of the organization."
344 )
345 socials: list[OrganizationSocialLink] | None = Field(
346 None, description="Links to social profiles."
347 )
348 details: OrganizationDetails | None = Field(
349 None,
350 description="Additional, private, business details Polar needs about active organizations for compliance (KYC).",
351 )
353 feature_settings: OrganizationFeatureSettings | None = None
354 subscription_settings: OrganizationSubscriptionSettings | None = None
355 notification_settings: OrganizationNotificationSettings | None = None
356 customer_email_settings: OrganizationCustomerEmailSettings | None = None
359class OrganizationPaymentStep(Schema):
360 id: str = Field(description="Step identifier")
361 title: str = Field(description="Step title")
362 description: str = Field(description="Step description")
363 completed: bool = Field(description="Whether the step is completed")
366class OrganizationPaymentStatus(Schema):
367 payment_ready: bool = Field(
368 description="Whether the organization is ready to accept payments"
369 )
370 steps: list[OrganizationPaymentStep] = Field(description="List of onboarding steps")
371 organization_status: OrganizationStatus = Field(
372 description="Current organization status"
373 )
376class OrganizationAppealRequest(Schema):
377 reason: Annotated[
378 str,
379 StringConstraints(min_length=50, max_length=5000),
380 Field(
381 description="Detailed explanation of why this organization should be approved. Minimum 50 characters."
382 ),
383 ]
386class OrganizationAppealResponse(Schema):
387 success: bool = Field(description="Whether the appeal was successfully submitted")
388 message: str = Field(description="Success or error message")
389 appeal_submitted_at: datetime = Field(description="When the appeal was submitted")
392class OrganizationReviewStatus(Schema):
393 verdict: Literal["PASS", "FAIL", "UNCERTAIN"] | None = Field(
394 default=None, description="AI validation verdict"
395 )
396 reason: str | None = Field(default=None, description="Reason for the verdict")
397 appeal_submitted_at: datetime | None = Field(
398 default=None, description="When appeal was submitted"
399 )
400 appeal_reason: str | None = Field(default=None, description="Reason for the appeal")
401 appeal_decision: OrganizationReview.AppealDecision | None = Field(
402 default=None, description="Decision on the appeal (approved/rejected)"
403 )
404 appeal_reviewed_at: datetime | None = Field(
405 default=None, description="When appeal was reviewed"
406 )
409class OrganizationDeletionBlockedReason(StrEnum):
410 """Reasons why an organization cannot be immediately deleted."""
412 HAS_ORDERS = "has_orders"
413 HAS_ACTIVE_SUBSCRIPTIONS = "has_active_subscriptions"
414 STRIPE_ACCOUNT_DELETION_FAILED = "stripe_account_deletion_failed"
417class OrganizationDeletionResponse(Schema):
418 """Response for organization deletion request."""
420 deleted: bool = Field(
421 description="Whether the organization was immediately deleted"
422 )
423 requires_support: bool = Field(
424 description="Whether a support ticket was created for manual handling"
425 )
426 blocked_reasons: list[OrganizationDeletionBlockedReason] = Field(
427 default_factory=list,
428 description="Reasons why immediate deletion is blocked",
429 )