Coverage for polar/customer_seat/schemas.py: 78%
79 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 12:42 +0000
1from datetime import datetime
2from typing import Any
3from uuid import UUID
5from pydantic import EmailStr, Field, field_validator, model_validator
6from sqlalchemy import inspect
8from polar.kit.schemas import Schema, TimestampedSchema
9from polar.models.customer_seat import SeatStatus
12class SeatAssign(Schema):
13 subscription_id: UUID | None = Field(
14 None,
15 description="Subscription ID. Required if checkout_id and order_id are not provided.",
16 )
17 checkout_id: UUID | None = Field(
18 None,
19 description="Checkout ID. Used to look up subscription or order from the checkout page.",
20 )
21 order_id: UUID | None = Field(
22 None,
23 description="Order ID for one-time purchases. Required if subscription_id and checkout_id are not provided.",
24 )
25 email: EmailStr | None = Field(
26 None, description="Email of the customer to assign the seat to"
27 )
28 external_customer_id: str | None = Field(
29 None, description="External customer ID for the seat assignment"
30 )
31 customer_id: UUID | None = Field(
32 None, description="Customer ID for the seat assignment"
33 )
34 metadata: dict[str, Any] | None = Field(
35 None, description="Additional metadata for the seat (max 10 keys, 1KB total)"
36 )
37 immediate_claim: bool = Field(
38 default=False,
39 description="If true, the seat will be immediately claimed without sending an invitation email. API-only feature.",
40 )
42 @field_validator("metadata")
43 @classmethod
44 def validate_metadata(cls, v: dict[str, Any] | None) -> dict[str, Any] | None:
45 if v is None:
46 return v
47 if len(v) > 10:
48 raise ValueError("Metadata cannot have more than 10 keys")
49 import json
51 if len(json.dumps(v)) > 1024: # 1KB limit 51 ↛ 52line 51 didn't jump to line 52 because the condition on line 51 was never true
52 raise ValueError("Metadata size cannot exceed 1KB")
53 return v
55 @model_validator(mode="after")
56 def validate_identifiers(self) -> "SeatAssign":
57 seat_source_identifiers = [
58 self.subscription_id,
59 self.checkout_id,
60 self.order_id,
61 ]
62 seat_source_count = sum(
63 1 for identifier in seat_source_identifiers if identifier is not None
64 )
66 if seat_source_count != 1:
67 raise ValueError(
68 "Exactly one of subscription_id, checkout_id, or order_id must be provided"
69 )
71 customer_identifiers = [self.email, self.external_customer_id, self.customer_id]
72 customer_count = sum(
73 1 for identifier in customer_identifiers if identifier is not None
74 )
76 if customer_count != 1:
77 raise ValueError(
78 "Exactly one of email, external_customer_id, or customer_id must be provided"
79 )
81 return self
84class SeatClaim(Schema):
85 invitation_token: str = Field(..., description="Invitation token to claim the seat")
88class CustomerSeat(TimestampedSchema):
89 id: UUID = Field(..., description="The seat ID")
90 subscription_id: UUID | None = Field(
91 None, description="The subscription ID (for recurring seats)"
92 )
93 order_id: UUID | None = Field(
94 None, description="The order ID (for one-time purchase seats)"
95 )
96 status: SeatStatus = Field(..., description="Status of the seat")
97 customer_id: UUID | None = Field(None, description="The assigned customer ID")
98 customer_email: str | None = Field(None, description="The assigned customer email")
99 invitation_token_expires_at: datetime | None = Field(
100 None, description="When the invitation token expires"
101 )
102 claimed_at: datetime | None = Field(None, description="When the seat was claimed")
103 revoked_at: datetime | None = Field(None, description="When the seat was revoked")
104 seat_metadata: dict[str, Any] | None = Field(
105 None, description="Additional metadata for the seat"
106 )
108 @model_validator(mode="before")
109 @classmethod
110 def extract_customer_email(cls, data: Any) -> Any:
111 if isinstance(data, dict):
112 # For dict data
113 if "customer" in data and data["customer"]:
114 data["customer_email"] = data.get("customer", {}).get("email")
115 return data
116 elif hasattr(data, "__dict__"):
117 # For SQLAlchemy models - check if customer is loaded
118 state = inspect(data)
119 if "customer" not in state.unloaded:
120 # Customer is loaded, we can extract the email
121 # But we need to let Pydantic handle the model conversion
122 # We'll just add the customer_email field if customer is available
123 if hasattr(data, "customer") and data.customer:
124 # Add customer_email as a temporary attribute
125 object.__setattr__(data, "customer_email", data.customer.email)
126 return data
129class SeatsList(Schema):
130 seats: list[CustomerSeat] = Field(..., description="List of seats")
131 available_seats: int = Field(..., description="Number of available seats")
132 total_seats: int = Field(
133 ..., description="Total number of seats for the subscription"
134 )
137class SeatClaimInfo(Schema):
138 """
139 Read-only information about a seat claim invitation.
140 Safe for email scanners - no side effects when fetched.
141 """
143 product_name: str = Field(..., description="Name of the product")
144 product_id: UUID = Field(..., description="ID of the product")
145 organization_name: str = Field(..., description="Name of the organization")
146 organization_slug: str = Field(..., description="Slug of the organization")
147 customer_email: str = Field(
148 ..., description="Email of the customer assigned to this seat"
149 )
150 can_claim: bool = Field(..., description="Whether the seat can be claimed")
153class CustomerSeatClaimResponse(Schema):
154 """Response after successfully claiming a seat."""
156 seat: CustomerSeat = Field(..., description="The claimed seat")
157 customer_session_token: str = Field(
158 ..., description="Session token for immediate customer portal access"
159 )
162CustomerSeatID = UUID