Coverage for polar/product/endpoints.py: 58%
47 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 15:52 +0000
1from typing import Annotated 1a
3from fastapi import Depends, Query 1a
5from polar.benefit.schemas import BenefitID 1a
6from polar.exceptions import NotPermitted, ResourceNotFound 1a
7from polar.kit.metadata import MetadataQuery, get_metadata_query_openapi_schema 1a
8from polar.kit.pagination import ListResource, PaginationParamsQuery 1a
9from polar.kit.schemas import MultipleQueryFilter 1a
10from polar.kit.sorting import Sorting, SortingGetter 1a
11from polar.models import Product 1a
12from polar.openapi import APITag 1a
13from polar.organization.schemas import OrganizationID 1a
14from polar.postgres import ( 1a
15 AsyncReadSession,
16 AsyncSession,
17 get_db_read_session,
18 get_db_session,
19)
20from polar.routing import APIRouter 1a
22from . import auth 1a
23from .schemas import Product as ProductSchema 1a
24from .schemas import ProductBenefitsUpdate, ProductCreate, ProductID, ProductUpdate 1a
25from .service import product as product_service 1a
26from .sorting import ProductSortProperty 1a
28router = APIRouter( 1a
29 prefix="/products",
30 tags=["products", APITag.public, APITag.mcp],
31)
33ProductNotFound = { 1a
34 "description": "Product not found.",
35 "model": ResourceNotFound.schema(),
36}
39ListSorting = Annotated[ 1a
40 list[Sorting[ProductSortProperty]],
41 Depends(SortingGetter(ProductSortProperty, ["-created_at"])),
42]
45@router.get( 1a
46 "/",
47 summary="List Products",
48 response_model=ListResource[ProductSchema],
49 openapi_extra={"parameters": [get_metadata_query_openapi_schema()]},
50)
51async def list( 1a
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 )
94 return ListResource.from_paginated_results(
95 [ProductSchema.model_validate(result) for result in results],
96 count,
97 pagination,
98 )
101@router.get( 1a
102 "/{id}",
103 summary="Get Product",
104 response_model=ProductSchema,
105 responses={404: ProductNotFound},
106)
107async def get( 1a
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)
115 if product is None:
116 raise ResourceNotFound()
118 return product
121@router.post( 1a
122 "/",
123 response_model=ProductSchema,
124 status_code=201,
125 summary="Create Product",
126 responses={201: {"description": "Product created."}},
127)
128async def create( 1a
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)
137@router.patch( 1a
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( 1a
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)
159 if product is None:
160 raise ResourceNotFound()
162 return await product_service.update(session, product, product_update, auth_subject)
165@router.post( 1a
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( 1a
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)
187 if product is None:
188 raise ResourceNotFound()
190 product, _, _ = await product_service.update_benefits(
191 session, product, benefits_update.benefits, auth_subject
192 )
193 return product