Coverage for polar/organization/schemas.py: 86%

173 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-12-05 17:15 +0000

1from datetime import datetime 1a

2from enum import StrEnum 1a

3from typing import Annotated, Any, Literal 1a

4 

5from pydantic import ( 1a

6 UUID4, 

7 AfterValidator, 

8 BeforeValidator, 

9 EmailStr, 

10 Field, 

11 StringConstraints, 

12 model_validator, 

13) 

14from pydantic.json_schema import SkipJsonSchema 1a

15from pydantic.networks import HttpUrl 1a

16 

17from polar.config import settings 1a

18from polar.enums import SubscriptionProrationBehavior 1a

19from polar.kit.email import EmailStrDNS 1a

20from polar.kit.schemas import ( 1a

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 ( 1a

32 OrganizationCustomerEmailSettings, 

33 OrganizationNotificationSettings, 

34 OrganizationStatus, 

35 OrganizationSubscriptionSettings, 

36) 

37from polar.models.organization_review import OrganizationReview 1a

38 

39OrganizationID = Annotated[ 1a

40 UUID4, 

41 MergeJSONSchema({"description": "The organization ID."}), 

42 SelectorWidget("/v1/organizations", "Organization", "name"), 

43 Field(examples=[ORGANIZATION_ID_EXAMPLE]), 

44] 

45 

46NameInput = Annotated[str, StringConstraints(min_length=3)] 1a

47 

48 

49def validate_reserved_keywords(value: str) -> str: 1a

50 if value in settings.ORGANIZATION_SLUG_RESERVED_KEYWORDS: 

51 raise ValueError("This slug is reserved.") 

52 return value 

53 

54 

55SlugInput = Annotated[ 1a

56 str, 

57 StringConstraints(to_lower=True, min_length=3), 

58 SlugValidator, 

59 AfterValidator(validate_reserved_keywords), 

60] 

61 

62 

63class OrganizationFeatureSettings(Schema): 1a

64 issue_funding_enabled: bool = Field( 1a

65 False, description="If this organization has issue funding enabled" 

66 ) 

67 seat_based_pricing_enabled: bool = Field( 1a

68 False, description="If this organization has seat-based pricing enabled" 

69 ) 

70 revops_enabled: bool = Field( 1a

71 False, description="If this organization has RevOps enabled" 

72 ) 

73 wallets_enabled: bool = Field( 1a

74 False, description="If this organization has Wallets enabled" 

75 ) 

76 member_model_enabled: bool = Field( 1a

77 False, description="If this organization has the Member model enabled" 

78 ) 

79 

80 

81class OrganizationSubscribePromoteSettings(Schema): 1a

82 promote: bool = Field(True, description="Promote email subscription (free)") 1a

83 show_count: bool = Field(True, description="Show subscription count publicly") 1a

84 count_free: bool = Field( 1a

85 True, description="Include free subscribers in total count" 

86 ) 

87 

88 

89class OrganizationDetails(Schema): 1a

90 about: str = Field( 1a

91 ..., description="Brief information about you and your business." 

92 ) 

93 product_description: str = Field( 1a

94 ..., description="Description of digital products being sold." 

95 ) 

96 intended_use: str = Field( 1a

97 ..., description="How the organization will integrate and use Polar." 

98 ) 

99 customer_acquisition: list[str] = Field( 1a

100 ..., description="Main customer acquisition channels." 

101 ) 

102 future_annual_revenue: int = Field( 1a

103 ..., ge=0, description="Estimated revenue in the next 12 months" 

104 ) 

105 switching: bool = Field(True, description="Switching from another platform?") 1a

106 switching_from: ( 1a

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( 1a

110 0, ge=0, description="Revenue from last year if applicable." 

111 ) 

112 

113 

114class OrganizationSocialPlatforms(StrEnum): 1a

115 x = "x" 1a

116 github = "github" 1a

117 facebook = "facebook" 1a

118 instagram = "instagram" 1a

119 youtube = "youtube" 1a

120 tiktok = "tiktok" 1a

121 linkedin = "linkedin" 1a

122 other = "other" 1a

123 

124 

125PLATFORM_DOMAINS = { 1a

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} 

134 

135 

136class OrganizationSocialLink(Schema): 1a

137 platform: OrganizationSocialPlatforms = Field( 1a

138 ..., description="The social platform of the URL" 

139 ) 

140 url: HttpUrlToStr = Field(..., description="The URL to the organization profile") 1a

141 

142 @model_validator(mode="before") 1a

143 @classmethod 1a

144 def validate_url(cls, data: dict[str, Any]) -> dict[str, Any]: 1a

145 platform = data.get("platform") 

146 url = data.get("url", "").lower() 

147 

148 if not (platform and url): 

149 return data 

150 

151 if platform == "other": 

152 return data 

153 

154 valid_domains = PLATFORM_DOMAINS[platform] 

155 if not any(domain in url for domain in valid_domains): 

156 raise ValueError( 

157 f"Invalid URL for {platform}. Must be from: {', '.join(valid_domains)}" 

158 ) 

159 

160 return data 

161 

162 

163# Deprecated 

164class OrganizationProfileSettings(Schema): 1a

165 enabled: bool | None = Field( 1a

166 None, description="If this organization has a profile enabled" 

167 ) 

168 description: Annotated[ 1a

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( 1a

174 None, description="A list of featured projects" 

175 ) 

176 featured_organizations: list[UUID4] | None = Field( 1a

177 None, description="A list of featured organizations" 

178 ) 

179 links: list[HttpUrl] | None = Field( 1a

180 None, description="A list of links associated with the organization" 

181 ) 

182 subscribe: OrganizationSubscribePromoteSettings | None = Field( 1a

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( 1a

191 None, description="Accent color for the organization" 

192 ) 

193 

194 

195class OrganizationBase(IDSchema, TimestampedSchema): 1a

196 name: str = Field( 1a

197 description="Organization name shown in checkout, customer portal, emails etc.", 

198 ) 

199 slug: str = Field( 1a

200 description="Unique organization slug in checkout, customer portal and credit card statements.", 

201 ) 

202 avatar_url: str | None = Field( 1a

203 description="Avatar URL shown in checkout, customer portal, emails etc." 

204 ) 

205 proration_behavior: SubscriptionProrationBehavior = Field( 1a

206 description="Proration behavior applied when customer updates their subscription from the portal.", 

207 ) 

208 allow_customer_updates: bool = Field( 1a

209 description="Whether customers can update their subscriptions from the customer portal.", 

210 ) 

211 

212 # Deprecated attributes 

213 bio: SkipJsonSchema[str | None] = Field(..., deprecated="") 1a

214 company: SkipJsonSchema[str | None] = Field( 1a

215 ..., 

216 deprecated="Legacy attribute no longer in use.", 

217 ) 

218 blog: SkipJsonSchema[str | None] = Field( 1a

219 ..., 

220 deprecated="Legacy attribute no longer in use. See `socials` instead.", 

221 ) 

222 location: SkipJsonSchema[str | None] = Field( 1a

223 ..., 

224 deprecated="Legacy attribute no longer in use.", 

225 ) 

226 twitter_username: SkipJsonSchema[str | None] = Field( 1a

227 ..., 

228 deprecated="Legacy attribute no longer in use. See `socials` instead.", 

229 ) 

230 

231 pledge_minimum_amount: SkipJsonSchema[int] = Field(0, deprecated=True) 1a

232 pledge_badge_show_amount: SkipJsonSchema[bool] = Field(False, deprecated=True) 1a

233 default_upfront_split_to_contributors: SkipJsonSchema[int | None] = Field( 1a

234 None, deprecated=True 

235 ) 

236 profile_settings: SkipJsonSchema[OrganizationProfileSettings | None] = Field( 1a

237 None, deprecated=True 

238 ) 

239 

240 

241class LegacyOrganizationStatus(StrEnum): 1a

242 """ 

243 Legacy organization status values kept for backward compatibility in schemas 

244 using OrganizationPublicBase. 

245 """ 

246 

247 CREATED = "created" 1a

248 ONBOARDING_STARTED = "onboarding_started" 1a

249 UNDER_REVIEW = "under_review" 1a

250 DENIED = "denied" 1a

251 ACTIVE = "active" 1a

252 

253 @classmethod 1a

254 def from_status(cls, status: OrganizationStatus) -> "LegacyOrganizationStatus": 1a

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 

269 

270 

271class OrganizationPublicBase(OrganizationBase): 1a

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] 1a

276 website: SkipJsonSchema[str | None] 1a

277 socials: SkipJsonSchema[list[OrganizationSocialLink]] 1a

278 status: Annotated[ 1a

279 SkipJsonSchema[LegacyOrganizationStatus], 

280 BeforeValidator(LegacyOrganizationStatus.from_status), 

281 ] 

282 details_submitted_at: SkipJsonSchema[datetime | None] 1a

283 

284 feature_settings: SkipJsonSchema[OrganizationFeatureSettings | None] 1a

285 subscription_settings: SkipJsonSchema[OrganizationSubscriptionSettings] 1a

286 notification_settings: SkipJsonSchema[OrganizationNotificationSettings] 1a

287 customer_email_settings: SkipJsonSchema[OrganizationCustomerEmailSettings] 1a

288 

289 

290class Organization(OrganizationBase): 1a

291 email: str | None = Field(description="Public support email.") 1a

292 website: str | None = Field(description="Official website of the organization.") 1a

293 socials: list[OrganizationSocialLink] = Field( 1a

294 description="Links to social profiles.", 

295 ) 

296 status: OrganizationStatus = Field(description="Current organization status") 1a

297 details_submitted_at: datetime | None = Field( 1a

298 description="When the business details were submitted.", 

299 ) 

300 

301 feature_settings: OrganizationFeatureSettings | None = Field( 1a

302 description="Organization feature settings", 

303 ) 

304 subscription_settings: OrganizationSubscriptionSettings = Field( 1a

305 description="Settings related to subscriptions management", 

306 ) 

307 notification_settings: OrganizationNotificationSettings = Field( 1a

308 description="Settings related to notifications", 

309 ) 

310 customer_email_settings: OrganizationCustomerEmailSettings = Field( 1a

311 description="Settings related to customer emails", 

312 ) 

313 

314 

315class OrganizationCreate(Schema): 1a

316 name: NameInput 1a

317 slug: SlugInput 1a

318 avatar_url: HttpUrlToStr | None = None 1a

319 email: EmailStr | None = Field(None, description="Public support email.") 1a

320 website: HttpUrlToStr | None = Field( 1a

321 None, description="Official website of the organization." 

322 ) 

323 socials: list[OrganizationSocialLink] | None = Field( 1a

324 None, 

325 description="Link to social profiles.", 

326 ) 

327 details: OrganizationDetails | None = Field( 1a

328 None, 

329 description="Additional, private, business details Polar needs about active organizations for compliance (KYC).", 

330 ) 

331 feature_settings: OrganizationFeatureSettings | None = None 1a

332 subscription_settings: OrganizationSubscriptionSettings | None = None 1a

333 notification_settings: OrganizationNotificationSettings | None = None 1a

334 customer_email_settings: OrganizationCustomerEmailSettings | None = None 1a

335 

336 

337class OrganizationUpdate(Schema): 1a

338 name: NameInput | None = None 1a

339 avatar_url: HttpUrlToStr | None = None 1a

340 

341 email: EmailStrDNS | None = Field(None, description="Public support email.") 1a

342 website: HttpUrlToStr | None = Field( 1a

343 None, description="Official website of the organization." 

344 ) 

345 socials: list[OrganizationSocialLink] | None = Field( 1a

346 None, description="Links to social profiles." 

347 ) 

348 details: OrganizationDetails | None = Field( 1a

349 None, 

350 description="Additional, private, business details Polar needs about active organizations for compliance (KYC).", 

351 ) 

352 

353 feature_settings: OrganizationFeatureSettings | None = None 1a

354 subscription_settings: OrganizationSubscriptionSettings | None = None 1a

355 notification_settings: OrganizationNotificationSettings | None = None 1a

356 customer_email_settings: OrganizationCustomerEmailSettings | None = None 1a

357 

358 

359class OrganizationPaymentStep(Schema): 1a

360 id: str = Field(description="Step identifier") 1a

361 title: str = Field(description="Step title") 1a

362 description: str = Field(description="Step description") 1a

363 completed: bool = Field(description="Whether the step is completed") 1a

364 

365 

366class OrganizationPaymentStatus(Schema): 1a

367 payment_ready: bool = Field( 1a

368 description="Whether the organization is ready to accept payments" 

369 ) 

370 steps: list[OrganizationPaymentStep] = Field(description="List of onboarding steps") 1a

371 organization_status: OrganizationStatus = Field( 1a

372 description="Current organization status" 

373 ) 

374 

375 

376class OrganizationAppealRequest(Schema): 1a

377 reason: Annotated[ 1a

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 ] 

384 

385 

386class OrganizationAppealResponse(Schema): 1a

387 success: bool = Field(description="Whether the appeal was successfully submitted") 1a

388 message: str = Field(description="Success or error message") 1a

389 appeal_submitted_at: datetime = Field(description="When the appeal was submitted") 1a

390 

391 

392class OrganizationReviewStatus(Schema): 1a

393 verdict: Literal["PASS", "FAIL", "UNCERTAIN"] | None = Field( 1a

394 default=None, description="AI validation verdict" 

395 ) 

396 reason: str | None = Field(default=None, description="Reason for the verdict") 1a

397 appeal_submitted_at: datetime | None = Field( 1a

398 default=None, description="When appeal was submitted" 

399 ) 

400 appeal_reason: str | None = Field(default=None, description="Reason for the appeal") 1a

401 appeal_decision: OrganizationReview.AppealDecision | None = Field( 1a

402 default=None, description="Decision on the appeal (approved/rejected)" 

403 ) 

404 appeal_reviewed_at: datetime | None = Field( 1a

405 default=None, description="When appeal was reviewed" 

406 ) 

407 

408 

409class OrganizationDeletionBlockedReason(StrEnum): 1a

410 """Reasons why an organization cannot be immediately deleted.""" 

411 

412 HAS_ORDERS = "has_orders" 1a

413 HAS_ACTIVE_SUBSCRIPTIONS = "has_active_subscriptions" 1a

414 STRIPE_ACCOUNT_DELETION_FAILED = "stripe_account_deletion_failed" 1a

415 

416 

417class OrganizationDeletionResponse(Schema): 1a

418 """Response for organization deletion request.""" 

419 

420 deleted: bool = Field( 1a

421 description="Whether the organization was immediately deleted" 

422 ) 

423 requires_support: bool = Field( 1a

424 description="Whether a support ticket was created for manual handling" 

425 ) 

426 blocked_reasons: list[OrganizationDeletionBlockedReason] = Field( 1a

427 default_factory=list, 

428 description="Reasons why immediate deletion is blocked", 

429 )