Coverage for polar/customer/service.py: 89%
123 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 12:42 +0000
1import uuid
2from collections.abc import Sequence
3from typing import Any
5from sqlalchemy import UnaryExpression, asc, desc, func, or_
6from sqlalchemy.orm import joinedload
7from sqlalchemy_utils.types.range import timedelta
9from polar.auth.models import AuthSubject
10from polar.benefit.grant.repository import BenefitGrantRepository
11from polar.customer_meter.repository import CustomerMeterRepository
12from polar.exceptions import PolarRequestValidationError, ValidationError
13from polar.kit.metadata import MetadataQuery, apply_metadata_clause
14from polar.kit.pagination import PaginationParams
15from polar.kit.sorting import Sorting
16from polar.member import member_service
17from polar.member.schemas import Member as MemberSchema
18from polar.models import BenefitGrant, Customer, Organization, User
19from polar.models.webhook_endpoint import CustomerWebhookEventType, WebhookEventType
20from polar.organization.resolver import get_payload_organization
21from polar.postgres import AsyncReadSession, AsyncSession
22from polar.redis import Redis
23from polar.subscription.repository import SubscriptionRepository
24from polar.webhook.service import webhook as webhook_service
25from polar.worker import enqueue_job
27from .repository import CustomerRepository
28from .schemas.customer import (
29 CustomerCreate,
30 CustomerUpdate,
31 CustomerUpdateExternalID,
32)
33from .schemas.state import CustomerState
34from .sorting import CustomerSortProperty
37class CustomerService:
38 async def list(
39 self,
40 session: AsyncReadSession,
41 auth_subject: AuthSubject[User | Organization],
42 *,
43 organization_id: Sequence[uuid.UUID] | None = None,
44 email: str | None = None,
45 metadata: MetadataQuery | None = None,
46 query: str | None = None,
47 pagination: PaginationParams,
48 sorting: list[Sorting[CustomerSortProperty]] = [
49 (CustomerSortProperty.created_at, True)
50 ],
51 ) -> tuple[Sequence[Customer], int]:
52 repository = CustomerRepository.from_session(session)
53 statement = repository.get_readable_statement(auth_subject)
55 if organization_id is not None: 55 ↛ 58line 55 didn't jump to line 58 because the condition on line 55 was always true
56 statement = statement.where(Customer.organization_id.in_(organization_id))
58 if email is not None:
59 statement = statement.where(func.lower(Customer.email) == email.lower())
61 if metadata is not None: 61 ↛ 64line 61 didn't jump to line 64 because the condition on line 61 was always true
62 statement = apply_metadata_clause(Customer, statement, metadata)
64 if query is not None:
65 statement = statement.where(
66 or_(
67 Customer.email.ilike(f"%{query}%"),
68 Customer.name.ilike(f"%{query}%"),
69 Customer.external_id.ilike(f"{query}%"),
70 )
71 )
73 order_by_clauses: list[UnaryExpression[Any]] = []
74 for criterion, is_desc in sorting:
75 clause_function = desc if is_desc else asc
76 if criterion == CustomerSortProperty.created_at:
77 order_by_clauses.append(clause_function(Customer.created_at))
78 elif criterion == CustomerSortProperty.email:
79 order_by_clauses.append(clause_function(Customer.email))
80 elif criterion == CustomerSortProperty.customer_name: 80 ↛ 74line 80 didn't jump to line 74 because the condition on line 80 was always true
81 order_by_clauses.append(clause_function(Customer.name))
82 statement = statement.order_by(*order_by_clauses)
84 return await repository.paginate(
85 statement, limit=pagination.limit, page=pagination.page
86 )
88 async def get(
89 self,
90 session: AsyncReadSession,
91 auth_subject: AuthSubject[User | Organization],
92 id: uuid.UUID,
93 ) -> Customer | None:
94 repository = CustomerRepository.from_session(session)
95 statement = repository.get_readable_statement(auth_subject).where(
96 Customer.id == id
97 )
98 return await repository.get_one_or_none(statement)
100 async def get_external(
101 self,
102 session: AsyncReadSession,
103 auth_subject: AuthSubject[User | Organization],
104 external_id: str,
105 ) -> Customer | None:
106 repository = CustomerRepository.from_session(session)
107 statement = repository.get_readable_statement(auth_subject).where(
108 Customer.external_id == external_id
109 )
110 return await repository.get_one_or_none(statement)
112 async def create(
113 self,
114 session: AsyncSession,
115 customer_create: CustomerCreate,
116 auth_subject: AuthSubject[User | Organization],
117 ) -> Customer:
118 organization = await get_payload_organization(
119 session, auth_subject, customer_create
120 )
121 repository = CustomerRepository.from_session(session)
123 errors: list[ValidationError] = []
125 if await repository.get_by_email_and_organization(
126 customer_create.email, organization.id
127 ):
128 errors.append(
129 {
130 "type": "value_error",
131 "loc": ("body", "email"),
132 "msg": "A customer with this email address already exists.",
133 "input": customer_create.email,
134 }
135 )
137 if customer_create.external_id is not None:
138 if await repository.get_by_external_id_and_organization(
139 customer_create.external_id, organization.id
140 ):
141 errors.append(
142 {
143 "type": "value_error",
144 "loc": ("body", "external_id"),
145 "msg": "A customer with this external ID already exists.",
146 "input": customer_create.external_id,
147 }
148 )
150 if errors:
151 raise PolarRequestValidationError(errors)
153 async with repository.create_context(
154 Customer(
155 organization=organization,
156 **customer_create.model_dump(
157 exclude={"organization_id", "owner"}, by_alias=True
158 ),
159 )
160 ) as customer:
161 owner_email = customer_create.owner.email if customer_create.owner else None
162 owner_name = customer_create.owner.name if customer_create.owner else None
163 owner_external_id = (
164 customer_create.owner.external_id if customer_create.owner else None
165 )
167 await member_service.create_owner_member(
168 session,
169 customer,
170 organization,
171 owner_email=owner_email,
172 owner_name=owner_name,
173 owner_external_id=owner_external_id,
174 )
175 return customer
177 async def update(
178 self,
179 session: AsyncSession,
180 customer: Customer,
181 customer_update: CustomerUpdate | CustomerUpdateExternalID,
182 ) -> Customer:
183 repository = CustomerRepository.from_session(session)
185 errors: list[ValidationError] = []
186 if (
187 customer_update.email is not None
188 and customer.email.lower() != customer_update.email.lower()
189 ):
190 already_exists = await repository.get_by_email_and_organization(
191 customer_update.email, customer.organization_id
192 )
193 if already_exists:
194 errors.append(
195 {
196 "type": "value_error",
197 "loc": ("body", "email"),
198 "msg": "A customer with this email address already exists.",
199 "input": customer_update.email,
200 }
201 )
203 customer.email = customer_update.email
204 customer.email_verified = False
206 if (
207 isinstance(customer_update, CustomerUpdate)
208 and "external_id" in customer_update.model_fields_set
209 and customer.external_id is not None
210 and customer.external_id != customer_update.external_id
211 ):
212 errors.append(
213 {
214 "type": "value_error",
215 "loc": ("body", "external_id"),
216 "msg": "Customer external ID cannot be updated.",
217 "input": customer_update.external_id,
218 }
219 )
221 if (
222 isinstance(customer_update, CustomerUpdate)
223 and customer_update.external_id is not None
224 and customer.external_id != customer_update.external_id
225 ):
226 if await repository.get_by_external_id_and_organization(
227 customer_update.external_id, customer.organization_id
228 ):
229 errors.append(
230 {
231 "type": "value_error",
232 "loc": ("body", "external_id"),
233 "msg": "A customer with this external ID already exists.",
234 "input": customer_update.external_id,
235 }
236 )
238 if errors:
239 raise PolarRequestValidationError(errors)
241 return await repository.update(
242 customer,
243 update_dict=customer_update.model_dump(
244 exclude={"email"}, exclude_unset=True, by_alias=True
245 ),
246 )
248 async def delete(self, session: AsyncSession, customer: Customer) -> Customer:
249 enqueue_job("subscription.cancel_customer", customer_id=customer.id)
250 enqueue_job("benefit.revoke_customer", customer_id=customer.id)
252 repository = CustomerRepository.from_session(session)
253 return await repository.soft_delete(customer)
255 async def get_state(
256 self,
257 session: AsyncReadSession,
258 redis: Redis,
259 customer: Customer,
260 cache: bool = True,
261 ) -> CustomerState:
262 # 👋 Whenever you change the state schema,
263 # please also update the cache key with a version number.
264 cache_key = f"polar:customer_state:v3:{customer.id}"
266 if cache: 266 ↛ 271line 266 didn't jump to line 271 because the condition on line 266 was always true
267 raw_state = await redis.get(cache_key)
268 if raw_state is not None:
269 return CustomerState.model_validate_json(raw_state)
271 subscription_repository = SubscriptionRepository.from_session(session)
272 customer.active_subscriptions = (
273 await subscription_repository.list_active_by_customer(customer.id)
274 )
276 benefit_grant_repository = BenefitGrantRepository.from_session(session)
277 customer.granted_benefits = (
278 await benefit_grant_repository.list_granted_by_customer(
279 customer.id, options=(joinedload(BenefitGrant.benefit),)
280 )
281 )
283 customer_meter_repository = CustomerMeterRepository.from_session(session)
284 customer.active_meters = await customer_meter_repository.get_all_by_customer(
285 customer.id
286 )
288 state = CustomerState.model_validate(customer)
290 await redis.set(
291 cache_key,
292 state.model_dump_json(),
293 ex=int(timedelta(hours=1).total_seconds()),
294 )
296 return state
298 async def webhook(
299 self,
300 session: AsyncSession,
301 redis: Redis,
302 event_type: CustomerWebhookEventType,
303 customer: Customer,
304 ) -> None:
305 if event_type == WebhookEventType.customer_state_changed:
306 data = await self.get_state(session, redis, customer, cache=False)
307 await webhook_service.send(
308 session,
309 customer.organization,
310 WebhookEventType.customer_state_changed,
311 data,
312 )
313 else:
314 await webhook_service.send(
315 session, customer.organization, event_type, customer
316 )
318 # For created, updated and deleted events, also trigger a state changed event
319 if event_type in (
320 WebhookEventType.customer_created,
321 WebhookEventType.customer_updated,
322 WebhookEventType.customer_deleted,
323 ):
324 await self.webhook(
325 session, redis, WebhookEventType.customer_state_changed, customer
326 )
328 async def load_members(
329 self,
330 session: AsyncReadSession,
331 customer_id: uuid.UUID,
332 ) -> Sequence[MemberSchema]:
333 members = await member_service.list_by_customer(session, customer_id)
334 return [MemberSchema.model_validate(member) for member in members]
337customer = CustomerService()