Coverage for polar/benefit/endpoints.py: 52%
53 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 17:15 +0000
1from fastapi import Depends, Query 1a
3from polar.customer.schemas.customer import CustomerID 1a
4from polar.exceptions import NotPermitted, ResourceNotFound 1a
5from polar.kit.metadata import MetadataQuery, get_metadata_query_openapi_schema 1a
6from polar.kit.pagination import ListResource, PaginationParamsQuery 1a
7from polar.kit.schemas import MultipleQueryFilter 1a
8from polar.models import Benefit 1a
9from polar.models.benefit import BenefitType 1a
10from polar.openapi import APITag 1a
11from polar.organization.schemas import OrganizationID 1a
12from polar.postgres import AsyncSession, get_db_session 1a
13from polar.redis import Redis, get_redis 1a
14from polar.routing import APIRouter 1a
16from . import auth, sorting 1a
17from .grant.service import benefit_grant as benefit_grant_service 1a
18from .schemas import Benefit as BenefitSchema 1a
19from .schemas import ( 1a
20 BenefitCreate,
21 BenefitGrant,
22 BenefitID,
23 BenefitUpdate,
24 benefit_schema_map,
25)
26from .service import benefit as benefit_service 1a
28router = APIRouter(prefix="/benefits", tags=["benefits", APITag.public]) 1a
30BenefitNotFound = { 1a
31 "description": "Benefit not found.",
32 "model": ResourceNotFound.schema(),
33}
36@router.get( 1a
37 "/",
38 summary="List Benefits",
39 response_model=ListResource[BenefitSchema],
40 openapi_extra={"parameters": [get_metadata_query_openapi_schema()]},
41)
42async def list( 1a
43 auth_subject: auth.BenefitsRead,
44 pagination: PaginationParamsQuery,
45 sorting: sorting.ListSorting,
46 metadata: MetadataQuery,
47 organization_id: MultipleQueryFilter[OrganizationID] | None = Query(
48 None, title="OrganizationID Filter", description="Filter by organization ID."
49 ),
50 type: MultipleQueryFilter[BenefitType] | None = Query(
51 None, title="BenefitType Filter", description="Filter by benefit type."
52 ),
53 session: AsyncSession = Depends(get_db_session),
54 query: str | None = Query(
55 None, title="Query", description="Filter by description."
56 ),
57) -> ListResource[BenefitSchema]:
58 """List benefits."""
59 results, count = await benefit_service.list(
60 session,
61 auth_subject,
62 type=type,
63 organization_id=organization_id,
64 metadata=metadata,
65 query=query,
66 pagination=pagination,
67 sorting=sorting,
68 )
70 return ListResource.from_paginated_results(
71 [benefit_schema_map[result.type].model_validate(result) for result in results],
72 count,
73 pagination,
74 )
77@router.get( 1a
78 "/{id}",
79 summary="Get Benefit",
80 response_model=BenefitSchema,
81 responses={404: BenefitNotFound},
82)
83async def get( 1a
84 id: BenefitID,
85 auth_subject: auth.BenefitsRead,
86 session: AsyncSession = Depends(get_db_session),
87) -> Benefit:
88 """Get a benefit by ID."""
89 benefit = await benefit_service.get(session, auth_subject, id)
91 if benefit is None:
92 raise ResourceNotFound()
94 return benefit
97@router.get( 1a
98 "/{id}/grants",
99 summary="List Benefit Grants",
100 response_model=ListResource[BenefitGrant],
101 responses={404: BenefitNotFound},
102)
103async def grants( 1a
104 id: BenefitID,
105 auth_subject: auth.BenefitsRead,
106 pagination: PaginationParamsQuery,
107 is_granted: bool | None = Query(
108 None,
109 description=(
110 "Filter by granted status. "
111 "If `true`, only granted benefits will be returned. "
112 "If `false`, only revoked benefits will be returned. "
113 ),
114 ),
115 customer_id: MultipleQueryFilter[CustomerID] | None = Query(
116 None, title="CustomerID Filter", description="Filter by customer."
117 ),
118 session: AsyncSession = Depends(get_db_session),
119) -> ListResource[BenefitGrant]:
120 """
121 List the individual grants for a benefit.
123 It's especially useful to check if a user has been granted a benefit.
124 """
125 benefit = await benefit_service.get(session, auth_subject, id)
127 if benefit is None:
128 raise ResourceNotFound()
130 results, count = await benefit_grant_service.list(
131 session,
132 benefit,
133 is_granted=is_granted,
134 customer_id=customer_id,
135 pagination=pagination,
136 )
138 return ListResource.from_paginated_results(
139 [BenefitGrant.model_validate(result) for result in results],
140 count,
141 pagination,
142 )
145@router.post( 1a
146 "/",
147 summary="Create Benefit",
148 response_model=BenefitSchema,
149 status_code=201,
150 responses={201: {"description": "Benefit created."}},
151)
152async def create( 1a
153 auth_subject: auth.BenefitsWrite,
154 benefit_create: BenefitCreate,
155 session: AsyncSession = Depends(get_db_session),
156 redis: Redis = Depends(get_redis),
157) -> Benefit:
158 """
159 Create a benefit.
160 """
161 benefit = await benefit_service.user_create(
162 session, redis, benefit_create, auth_subject
163 )
165 return benefit
168@router.patch( 1a
169 "/{id}",
170 summary="Update Benefit",
171 response_model=BenefitSchema,
172 responses={
173 200: {"description": "Benefit updated."},
174 404: BenefitNotFound,
175 },
176)
177async def update( 1a
178 id: BenefitID,
179 benefit_update: BenefitUpdate,
180 auth_subject: auth.BenefitsWrite,
181 session: AsyncSession = Depends(get_db_session),
182 redis: Redis = Depends(get_redis),
183) -> Benefit:
184 """
185 Update a benefit.
186 """
187 benefit = await benefit_service.get(session, auth_subject, id)
189 if benefit is None:
190 raise ResourceNotFound()
192 return await benefit_service.update(
193 session, redis, benefit, benefit_update, auth_subject
194 )
197@router.delete( 1a
198 "/{id}",
199 summary="Delete Benefit",
200 status_code=204,
201 responses={
202 204: {"description": "Benefit deleted."},
203 403: {
204 "description": "This benefit is not deletable.",
205 "model": NotPermitted.schema(),
206 },
207 404: BenefitNotFound,
208 },
209)
210async def delete( 1a
211 id: BenefitID,
212 auth_subject: auth.BenefitsWrite,
213 session: AsyncSession = Depends(get_db_session),
214) -> None:
215 """
216 Delete a benefit.
218 > [!WARNING]
219 > Every grants associated with the benefit will be revoked.
220 > Users will lose access to the benefit.
221 """
222 benefit = await benefit_service.get(session, auth_subject, id)
224 if benefit is None:
225 raise ResourceNotFound()
227 await benefit_service.delete(session, benefit)