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

1from typing import Annotated 

2 

3from fastapi import Depends, Query 

4 

5from polar.benefit.schemas import BenefitID 

6from polar.exceptions import NotPermitted, ResourceNotFound 

7from polar.kit.metadata import MetadataQuery, get_metadata_query_openapi_schema 

8from polar.kit.pagination import ListResource, PaginationParamsQuery 

9from polar.kit.schemas import MultipleQueryFilter 

10from polar.kit.sorting import Sorting, SortingGetter 

11from polar.models import Product 

12from polar.openapi import APITag 

13from polar.organization.schemas import OrganizationID 

14from polar.postgres import ( 

15 AsyncReadSession, 

16 AsyncSession, 

17 get_db_read_session, 

18 get_db_session, 

19) 

20from polar.routing import APIRouter 

21 

22from . import auth 

23from .schemas import Product as ProductSchema 

24from .schemas import ProductBenefitsUpdate, ProductCreate, ProductID, ProductUpdate 

25from .service import product as product_service 

26from .sorting import ProductSortProperty 

27 

28router = APIRouter( 

29 prefix="/products", 

30 tags=["products", APITag.public, APITag.mcp], 

31) 

32 

33ProductNotFound = { 

34 "description": "Product not found.", 

35 "model": ResourceNotFound.schema(), 

36} 

37 

38 

39ListSorting = Annotated[ 

40 list[Sorting[ProductSortProperty]], 

41 Depends(SortingGetter(ProductSortProperty, ["-created_at"])), 

42] 

43 

44 

45@router.get( 

46 "/", 

47 summary="List Products", 

48 response_model=ListResource[ProductSchema], 

49 openapi_extra={"parameters": [get_metadata_query_openapi_schema()]}, 

50) 

51async def list( 

52 pagination: PaginationParamsQuery, 

53 sorting: ListSorting, 

54 auth_subject: auth.CreatorProductsRead, 

55 metadata: MetadataQuery, 

56 id: MultipleQueryFilter[ProductID] | None = Query( 

57 None, title="ProductID Filter", description="Filter by product ID." 

58 ), 

59 organization_id: MultipleQueryFilter[OrganizationID] | None = Query( 

60 None, title="OrganizationID Filter", description="Filter by organization ID." 

61 ), 

62 query: str | None = Query(None, description="Filter by product name."), 

63 is_archived: bool | None = Query(None, description="Filter on archived products."), 

64 is_recurring: bool | None = Query( 

65 None, 

66 description=( 

67 "Filter on recurring products. " 

68 "If `true`, only subscriptions tiers are returned. " 

69 "If `false`, only one-time purchase products are returned. " 

70 ), 

71 ), 

72 benefit_id: MultipleQueryFilter[BenefitID] | None = Query( 

73 None, 

74 title="BenefitID Filter", 

75 description="Filter products granting specific benefit.", 

76 ), 

77 session: AsyncReadSession = Depends(get_db_read_session), 

78) -> ListResource[ProductSchema]: 

79 """List products.""" 

80 results, count = await product_service.list( 

81 session, 

82 auth_subject, 

83 id=id, 

84 organization_id=organization_id, 

85 query=query, 

86 is_archived=is_archived, 

87 is_recurring=is_recurring, 

88 benefit_id=benefit_id, 

89 metadata=metadata, 

90 pagination=pagination, 

91 sorting=sorting, 

92 ) 

93 

94 return ListResource.from_paginated_results( 

95 [ProductSchema.model_validate(result) for result in results], 

96 count, 

97 pagination, 

98 ) 

99 

100 

101@router.get( 

102 "/{id}", 

103 summary="Get Product", 

104 response_model=ProductSchema, 

105 responses={404: ProductNotFound}, 

106) 

107async def get( 

108 id: ProductID, 

109 auth_subject: auth.CreatorProductsRead, 

110 session: AsyncReadSession = Depends(get_db_read_session), 

111) -> Product: 

112 """Get a product by ID.""" 

113 product = await product_service.get(session, auth_subject, id) 

114 

115 if product is None: 

116 raise ResourceNotFound() 

117 

118 return product 

119 

120 

121@router.post( 

122 "/", 

123 response_model=ProductSchema, 

124 status_code=201, 

125 summary="Create Product", 

126 responses={201: {"description": "Product created."}}, 

127) 

128async def create( 

129 product_create: ProductCreate, 

130 auth_subject: auth.CreatorProductsWrite, 

131 session: AsyncSession = Depends(get_db_session), 

132) -> Product: 

133 """Create a product.""" 

134 return await product_service.create(session, product_create, auth_subject) 

135 

136 

137@router.patch( 

138 "/{id}", 

139 response_model=ProductSchema, 

140 summary="Update Product", 

141 responses={ 

142 200: {"description": "Product updated."}, 

143 403: { 

144 "description": "You don't have the permission to update this product.", 

145 "model": NotPermitted.schema(), 

146 }, 

147 404: ProductNotFound, 

148 }, 

149) 

150async def update( 

151 id: ProductID, 

152 product_update: ProductUpdate, 

153 auth_subject: auth.CreatorProductsWrite, 

154 session: AsyncSession = Depends(get_db_session), 

155) -> Product: 

156 """Update a product.""" 

157 product = await product_service.get(session, auth_subject, id) 

158 

159 if product is None: 

160 raise ResourceNotFound() 

161 

162 return await product_service.update(session, product, product_update, auth_subject) 

163 

164 

165@router.post( 

166 "/{id}/benefits", 

167 response_model=ProductSchema, 

168 summary="Update Product Benefits", 

169 responses={ 

170 200: {"description": "Product benefits updated."}, 

171 403: { 

172 "description": "You don't have the permission to update this product.", 

173 "model": NotPermitted.schema(), 

174 }, 

175 404: ProductNotFound, 

176 }, 

177) 

178async def update_benefits( 

179 id: ProductID, 

180 benefits_update: ProductBenefitsUpdate, 

181 auth_subject: auth.CreatorProductsWrite, 

182 session: AsyncSession = Depends(get_db_session), 

183) -> Product: 

184 """Update benefits granted by a product.""" 

185 product = await product_service.get(session, auth_subject, id) 

186 

187 if product is None: 

188 raise ResourceNotFound() 

189 

190 product, _, _ = await product_service.update_benefits( 

191 session, product, benefits_update.benefits, auth_subject 

192 ) 

193 return product