Coverage for polar/custom_field/endpoints.py: 58%
44 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 16:17 +0000
1from typing import Annotated 1a
3from fastapi import Depends, Path, Query 1a
4from pydantic import UUID4 1a
6from polar.exceptions import ResourceNotFound 1a
7from polar.kit.pagination import ListResource, PaginationParamsQuery 1a
8from polar.kit.schemas import MultipleQueryFilter 1a
9from polar.models import CustomField 1a
10from polar.models.custom_field import CustomFieldType 1a
11from polar.openapi import APITag 1a
12from polar.organization.schemas import OrganizationID 1a
13from polar.postgres import ( 1a
14 AsyncReadSession,
15 AsyncSession,
16 get_db_read_session,
17 get_db_session,
18)
19from polar.routing import APIRouter 1a
21from . import auth, sorting 1a
22from .schemas import CustomField as CustomFieldSchema 1a
23from .schemas import CustomFieldAdapter, CustomFieldCreate, CustomFieldUpdate 1a
24from .service import custom_field as custom_field_service 1a
26router = APIRouter(prefix="/custom-fields", tags=["custom-fields", APITag.public]) 1a
29CustomFieldID = Annotated[UUID4, Path(description="The custom field ID.")] 1a
30CustomFieldNotFound = { 1a
31 "description": "Custom field not found.",
32 "model": ResourceNotFound.schema(),
33}
36@router.get( 1a
37 "/", summary="List Custom Fields", response_model=ListResource[CustomFieldSchema]
38)
39async def list( 1a
40 auth_subject: auth.CustomFieldRead,
41 pagination: PaginationParamsQuery,
42 sorting: sorting.ListSorting,
43 organization_id: MultipleQueryFilter[OrganizationID] | None = Query(
44 None, title="OrganizationID Filter", description="Filter by organization ID."
45 ),
46 query: str | None = Query(None, description="Filter by custom field name or slug."),
47 type: MultipleQueryFilter[CustomFieldType] | None = Query(
48 None, title="CustomFieldType Filter", description="Filter by custom field type."
49 ),
50 session: AsyncReadSession = Depends(get_db_read_session),
51) -> ListResource[CustomFieldSchema]:
52 """List custom fields."""
53 results, count = await custom_field_service.list(
54 session,
55 auth_subject,
56 organization_id=organization_id,
57 query=query,
58 type=type,
59 pagination=pagination,
60 sorting=sorting,
61 )
63 return ListResource.from_paginated_results(
64 [CustomFieldAdapter.validate_python(result) for result in results],
65 count,
66 pagination,
67 )
70@router.get( 1a
71 "/{id}",
72 summary="Get Custom Field",
73 response_model=CustomFieldSchema,
74 responses={404: CustomFieldNotFound},
75)
76async def get( 1a
77 id: CustomFieldID,
78 auth_subject: auth.CustomFieldRead,
79 session: AsyncReadSession = Depends(get_db_read_session),
80) -> CustomField:
81 """Get a custom field by ID."""
82 custom_field = await custom_field_service.get_by_id(session, auth_subject, id)
84 if custom_field is None:
85 raise ResourceNotFound()
87 return custom_field
90@router.post( 1a
91 "/",
92 response_model=CustomFieldSchema,
93 status_code=201,
94 summary="Create Custom Field",
95 responses={201: {"description": "Custom field created."}},
96)
97async def create( 1a
98 custom_field_create: CustomFieldCreate,
99 auth_subject: auth.CustomFieldWrite,
100 session: AsyncSession = Depends(get_db_session),
101) -> CustomField:
102 """Create a custom field."""
103 return await custom_field_service.create(session, custom_field_create, auth_subject)
106@router.patch( 1a
107 "/{id}",
108 response_model=CustomFieldSchema,
109 summary="Update Custom Field",
110 responses={
111 200: {"description": "Custom field updated."},
112 404: CustomFieldNotFound,
113 },
114)
115async def update( 1a
116 id: CustomFieldID,
117 custom_field_update: CustomFieldUpdate,
118 auth_subject: auth.CustomFieldWrite,
119 session: AsyncSession = Depends(get_db_session),
120) -> CustomField:
121 """Update a custom field."""
122 custom_field = await custom_field_service.get_by_id(session, auth_subject, id)
124 if custom_field is None:
125 raise ResourceNotFound()
127 return await custom_field_service.update(session, custom_field, custom_field_update)
130@router.delete( 1a
131 "/{id}",
132 status_code=204,
133 summary="Delete Custom Field",
134 responses={
135 204: {"description": "Custom field deleted."},
136 404: CustomFieldNotFound,
137 },
138)
139async def delete( 1a
140 id: CustomFieldID,
141 auth_subject: auth.CustomFieldWrite,
142 session: AsyncSession = Depends(get_db_session),
143) -> None:
144 """Delete a custom field."""
145 custom_field = await custom_field_service.get_by_id(session, auth_subject, id)
147 if custom_field is None:
148 raise ResourceNotFound()
150 await custom_field_service.delete(session, custom_field)