Coverage for polar/custom_field/service.py: 94%
93 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 (
6 Select,
7 UnaryExpression,
8 asc,
9 delete,
10 desc,
11 func,
12 or_,
13 select,
14 update,
15)
16from sqlalchemy.orm import contains_eager
18from polar.auth.models import AuthSubject, is_organization, is_user
19from polar.custom_field.sorting import CustomFieldSortProperty
20from polar.exceptions import PolarError, PolarRequestValidationError
21from polar.kit.pagination import PaginationParams, paginate
22from polar.kit.services import ResourceServiceReader
23from polar.kit.sorting import Sorting
24from polar.models import CustomField, Organization, User, UserOrganization
25from polar.models.custom_field import CustomFieldType
26from polar.organization.resolver import get_payload_organization
27from polar.postgres import AsyncReadSession, AsyncSession
29from .attachment import attached_custom_fields_models
30from .data import custom_field_data_models
31from .schemas import CustomFieldCreate, CustomFieldUpdate
34class CustomFieldError(PolarError): ... 34 ↛ 37line 34 didn't jump to line 37 because
37class CustomFieldService(ResourceServiceReader[CustomField]):
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 query: str | None = None,
45 type: Sequence[CustomFieldType] | None = None,
46 pagination: PaginationParams,
47 sorting: list[Sorting[CustomFieldSortProperty]] = [
48 (CustomFieldSortProperty.slug, False)
49 ],
50 ) -> tuple[Sequence[CustomField], int]:
51 statement = self._get_readable_custom_field_statement(auth_subject)
53 if organization_id is not None: 53 ↛ 58line 53 didn't jump to line 58 because the condition on line 53 was always true
54 statement = statement.where(
55 CustomField.organization_id.in_(organization_id)
56 )
58 if query is not None:
59 statement = statement.where(
60 or_(
61 CustomField.name.ilike(f"%{query}%"),
62 CustomField.slug.ilike(f"%{query}%"),
63 )
64 )
66 if type is not None:
67 statement = statement.where(CustomField.type.in_(type))
69 order_by_clauses: list[UnaryExpression[Any]] = []
70 for criterion, is_desc in sorting:
71 clause_function = desc if is_desc else asc
72 if criterion == CustomFieldSortProperty.created_at:
73 order_by_clauses.append(
74 clause_function(CustomFieldSortProperty.created_at)
75 )
76 elif criterion == CustomFieldSortProperty.slug:
77 order_by_clauses.append(clause_function(CustomFieldSortProperty.slug))
78 elif criterion == CustomFieldSortProperty.custom_field_name:
79 order_by_clauses.append(
80 clause_function(CustomFieldSortProperty.custom_field_name)
81 )
82 elif criterion == CustomFieldSortProperty.type: 82 ↛ 70line 82 didn't jump to line 70 because the condition on line 82 was always true
83 order_by_clauses.append(clause_function(CustomFieldSortProperty.type))
84 statement = statement.order_by(*order_by_clauses)
86 return await paginate(session, statement, pagination=pagination)
88 async def get_by_id(
89 self,
90 session: AsyncReadSession,
91 auth_subject: AuthSubject[User | Organization],
92 id: uuid.UUID,
93 ) -> CustomField | None:
94 statement = self._get_readable_custom_field_statement(auth_subject).where(
95 CustomField.id == id
96 )
97 result = await session.execute(statement)
98 return result.scalar_one_or_none()
100 async def create(
101 self,
102 session: AsyncSession,
103 custom_field_create: CustomFieldCreate,
104 auth_subject: AuthSubject[User | Organization],
105 ) -> CustomField:
106 organization = await get_payload_organization(
107 session, auth_subject, custom_field_create
108 )
110 existing_field = await self._get_by_organization_id_and_slug(
111 session, organization.id, custom_field_create.slug
112 )
113 if existing_field is not None:
114 raise PolarRequestValidationError(
115 [
116 {
117 "type": "value_error",
118 "loc": ("body", "slug"),
119 "msg": "Custom field with this slug already exists.",
120 "input": custom_field_create.slug,
121 }
122 ]
123 )
125 custom_field = CustomField(
126 **custom_field_create.model_dump(
127 exclude={"organization_id"}, by_alias=True
128 ),
129 organization=organization,
130 )
131 session.add(custom_field)
133 return custom_field
135 async def update(
136 self,
137 session: AsyncSession,
138 custom_field: CustomField,
139 custom_field_update: CustomFieldUpdate,
140 ) -> CustomField:
141 if custom_field.type != custom_field_update.type:
142 raise PolarRequestValidationError(
143 [
144 {
145 "type": "value_error",
146 "loc": ("body", "type"),
147 "msg": "The type of a custom field can't be changed.",
148 "input": custom_field_update.type,
149 }
150 ]
151 )
153 if (
154 custom_field_update.slug is not None
155 and custom_field.slug != custom_field_update.slug
156 ):
157 existing_field = await self._get_by_organization_id_and_slug(
158 session, custom_field.organization_id, custom_field_update.slug
159 )
160 if existing_field is not None and existing_field.id != custom_field.id:
161 raise PolarRequestValidationError(
162 [
163 {
164 "type": "value_error",
165 "loc": ("body", "slug"),
166 "msg": "Custom field with this slug already exists.",
167 "input": custom_field_update.slug,
168 }
169 ]
170 )
172 previous_slug = custom_field.slug
173 for attr, value in custom_field_update.model_dump(
174 exclude_unset=True, by_alias=True
175 ).items():
176 setattr(custom_field, attr, value)
178 # Update the slug from all custom_field_data JSONB
179 if previous_slug != custom_field.slug:
180 for model in custom_field_data_models:
181 update_statement = (
182 update(model)
183 .where(
184 model.organization == custom_field.organization,
185 model.custom_field_data.has_key(previous_slug),
186 )
187 .values(
188 custom_field_data=(
189 model.custom_field_data.op("-")(previous_slug)
190 ).op("||")(
191 func.jsonb_build_object(
192 custom_field.slug,
193 model.custom_field_data[previous_slug],
194 )
195 )
196 )
197 )
198 await session.execute(update_statement)
200 session.add(custom_field)
201 return custom_field
203 async def delete(
204 self, session: AsyncSession, custom_field: CustomField
205 ) -> CustomField:
206 custom_field.set_deleted_at()
207 session.add(custom_field)
209 # Delete row with this custom field from all association tables
210 for model in attached_custom_fields_models:
211 delete_statement = delete(model).where(
212 model.custom_field_id == custom_field.id
213 )
214 await session.execute(delete_statement)
216 return custom_field
218 async def get_by_organization_and_id(
219 self, session: AsyncSession, id: uuid.UUID, organization_id: uuid.UUID
220 ) -> CustomField | None:
221 statement = select(CustomField).where(
222 CustomField.deleted_at.is_(None),
223 CustomField.organization_id == organization_id,
224 CustomField.id == id,
225 )
226 result = await session.execute(statement)
227 return result.scalar_one_or_none()
229 async def _get_by_organization_id_and_slug(
230 self, session: AsyncSession, organization_id: uuid.UUID, slug: str
231 ) -> CustomField | None:
232 statement = select(CustomField).where(
233 CustomField.organization_id == organization_id,
234 CustomField.slug == slug,
235 )
236 result = await session.execute(statement)
237 return result.scalar_one_or_none()
239 def _get_readable_custom_field_statement(
240 self, auth_subject: AuthSubject[User | Organization]
241 ) -> Select[tuple[CustomField]]:
242 statement = (
243 select(CustomField)
244 .where(CustomField.deleted_at.is_(None))
245 .join(Organization, Organization.id == CustomField.organization_id)
246 .options(contains_eager(CustomField.organization))
247 )
249 if is_user(auth_subject): 249 ↛ 259line 249 didn't jump to line 259 because the condition on line 249 was always true
250 user = auth_subject.subject
251 statement = statement.where(
252 CustomField.organization_id.in_(
253 select(UserOrganization.organization_id).where(
254 UserOrganization.user_id == user.id,
255 UserOrganization.deleted_at.is_(None),
256 )
257 )
258 )
259 elif is_organization(auth_subject):
260 statement = statement.where(
261 CustomField.organization_id == auth_subject.subject.id,
262 )
264 return statement
267custom_field = CustomFieldService(CustomField)