Coverage for polar/custom_field/endpoints.py: 100%

1from typing import Annotated 

2 

3from fastapi import Depends, Path, Query 

4from pydantic import UUID4 

5 

6from polar.exceptions import ResourceNotFound 

7from polar.kit.pagination import ListResource, PaginationParamsQuery 

8from polar.kit.schemas import MultipleQueryFilter 

9from polar.models import CustomField 

10from polar.models.custom_field import CustomFieldType 

11from polar.openapi import APITag 

12from polar.organization.schemas import OrganizationID 

13from polar.postgres import ( 

14 AsyncReadSession, 

15 AsyncSession, 

16 get_db_read_session, 

17 get_db_session, 

18) 

19from polar.routing import APIRouter 

20 

21from . import auth, sorting 

22from .schemas import CustomField as CustomFieldSchema 

23from .schemas import CustomFieldAdapter, CustomFieldCreate, CustomFieldUpdate 

24from .service import custom_field as custom_field_service 

25 

26router = APIRouter(prefix="/custom-fields", tags=["custom-fields", APITag.public]) 

27 

28 

29CustomFieldID = Annotated[UUID4, Path(description="The custom field ID.")] 

30CustomFieldNotFound = { 

31 "description": "Custom field not found.", 

32 "model": ResourceNotFound.schema(), 

33} 

34 

35 

36@router.get( 

37 "/", summary="List Custom Fields", response_model=ListResource[CustomFieldSchema] 

38) 

39async def list( 

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 ) 

62 

63 return ListResource.from_paginated_results( 

64 [CustomFieldAdapter.validate_python(result) for result in results], 

65 count, 

66 pagination, 

67 ) 

68 

69 

70@router.get( 

71 "/{id}", 

72 summary="Get Custom Field", 

73 response_model=CustomFieldSchema, 

74 responses={404: CustomFieldNotFound}, 

75) 

76async def get( 

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) 

83 

84 if custom_field is None: 

85 raise ResourceNotFound() 

86 

87 return custom_field 

88 

89 

90@router.post( 

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( 

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) 

104 

105 

106@router.patch( 

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( 

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) 

123 

124 if custom_field is None: 

125 raise ResourceNotFound() 

126 

127 return await custom_field_service.update(session, custom_field, custom_field_update) 

128 

129 

130@router.delete( 

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( 

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) 

146 

147 if custom_field is None: 

148 raise ResourceNotFound() 

149 

150 await custom_field_service.delete(session, custom_field)