Coverage for polar/customer_portal/endpoints/benefit_grant.py: 67%
39 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 NotPermitted, ResourceNotFound 1a
7from polar.kit.db.postgres import AsyncSession 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 BenefitGrant 1a
12from polar.models.benefit import BenefitType 1a
13from polar.openapi import APITag 1a
14from polar.order.schemas import OrderID 1a
15from polar.postgres import get_db_session 1a
16from polar.routing import APIRouter 1a
17from polar.subscription.schemas import SubscriptionID 1a
19from .. import auth 1a
20from ..schemas.benefit_grant import ( 1a
21 CustomerBenefitGrant,
22 CustomerBenefitGrantAdapter,
23 CustomerBenefitGrantUpdate,
24)
25from ..service.benefit_grant import CustomerBenefitGrantSortProperty 1a
26from ..service.benefit_grant import ( 1a
27 customer_benefit_grant as customer_benefit_grant_service,
28)
30router = APIRouter( 1a
31 prefix="/benefit-grants",
32 tags=["benefit-grants", APITag.public],
33)
35BenefitGrantID = Annotated[UUID4, Path(description="The benefit grant ID.")] 1a
36BenefitGrantNotFound = { 1a
37 "description": "Benefit grant not found.",
38 "model": ResourceNotFound.schema(),
39}
41ListSorting = Annotated[ 1a
42 list[Sorting[CustomerBenefitGrantSortProperty]],
43 Depends(
44 SortingGetter(
45 CustomerBenefitGrantSortProperty, ["product_benefit", "-granted_at"]
46 )
47 ),
48]
51@router.get( 1a
52 "/",
53 summary="List Benefit Grants",
54 response_model=ListResource[CustomerBenefitGrant],
55)
56async def list( 1a
57 auth_subject: auth.CustomerPortalRead,
58 pagination: PaginationParamsQuery,
59 sorting: ListSorting,
60 type: MultipleQueryFilter[BenefitType] | None = Query(
61 None, title="BenefitType Filter", description="Filter by benefit type."
62 ),
63 benefit_id: MultipleQueryFilter[UUID4] | None = Query(
64 None, title="BenefitID Filter", description="Filter by benefit ID."
65 ),
66 checkout_id: MultipleQueryFilter[UUID4] | None = Query(
67 None, title="CheckoutID Filter", description="Filter by checkout ID."
68 ),
69 order_id: MultipleQueryFilter[OrderID] | None = Query(
70 None, title="OrderID Filter", description="Filter by order ID."
71 ),
72 subscription_id: MultipleQueryFilter[SubscriptionID] | None = Query(
73 None, title="SubscriptionID Filter", description="Filter by subscription ID."
74 ),
75 session: AsyncSession = Depends(get_db_session),
76) -> ListResource[CustomerBenefitGrant]:
77 """List benefits grants of the authenticated customer."""
78 results, count = await customer_benefit_grant_service.list(
79 session,
80 auth_subject,
81 type=type,
82 benefit_id=benefit_id,
83 checkout_id=checkout_id,
84 order_id=order_id,
85 subscription_id=subscription_id,
86 pagination=pagination,
87 sorting=sorting,
88 )
90 return ListResource.from_paginated_results(
91 [CustomerBenefitGrantAdapter.validate_python(result) for result in results],
92 count,
93 pagination,
94 )
97@router.get( 1a
98 "/{id}",
99 summary="Get Benefit Grant",
100 response_model=CustomerBenefitGrant,
101 responses={404: BenefitGrantNotFound},
102)
103async def get( 1a
104 id: BenefitGrantID,
105 auth_subject: auth.CustomerPortalRead,
106 session: AsyncSession = Depends(get_db_session),
107) -> BenefitGrant:
108 """Get a benefit grant by ID for the authenticated customer."""
109 benefit_grant = await customer_benefit_grant_service.get_by_id(
110 session, auth_subject, id
111 )
113 if benefit_grant is None:
114 raise ResourceNotFound()
116 return benefit_grant
119@router.patch( 1a
120 "/{id}",
121 summary="Update Benefit Grant",
122 response_model=CustomerBenefitGrant,
123 responses={
124 200: {"description": "Benefit grant updated."},
125 403: {
126 "description": "The benefit grant is revoked and cannot be updated.",
127 "model": NotPermitted.schema(),
128 },
129 404: BenefitGrantNotFound,
130 },
131)
132async def update( 1a
133 id: BenefitGrantID,
134 benefit_grant_update: CustomerBenefitGrantUpdate,
135 auth_subject: auth.CustomerPortalWrite,
136 session: AsyncSession = Depends(get_db_session),
137) -> BenefitGrant:
138 """Update a benefit grant for the authenticated customer."""
139 benefit_grant = await customer_benefit_grant_service.get_by_id(
140 session, auth_subject, id
141 )
143 if benefit_grant is None:
144 raise ResourceNotFound()
146 return await customer_benefit_grant_service.update(
147 session, benefit_grant, benefit_grant_update
148 )