Coverage for polar/customer_portal/endpoints/benefit_grant.py: 91%

1from typing import Annotated 

2 

3from fastapi import Depends, Path, Query 

4from pydantic import UUID4 

5 

6from polar.exceptions import NotPermitted, ResourceNotFound 

7from polar.kit.db.postgres import AsyncSession 

8from polar.kit.pagination import ListResource, PaginationParamsQuery 

9from polar.kit.schemas import MultipleQueryFilter 

10from polar.kit.sorting import Sorting, SortingGetter 

11from polar.models import BenefitGrant 

12from polar.models.benefit import BenefitType 

13from polar.openapi import APITag 

14from polar.order.schemas import OrderID 

15from polar.postgres import get_db_session 

16from polar.routing import APIRouter 

17from polar.subscription.schemas import SubscriptionID 

18 

19from .. import auth 

20from ..schemas.benefit_grant import ( 

21 CustomerBenefitGrant, 

22 CustomerBenefitGrantAdapter, 

23 CustomerBenefitGrantUpdate, 

24) 

25from ..service.benefit_grant import CustomerBenefitGrantSortProperty 

26from ..service.benefit_grant import ( 

27 customer_benefit_grant as customer_benefit_grant_service, 

28) 

29 

30router = APIRouter( 

31 prefix="/benefit-grants", 

32 tags=["benefit-grants", APITag.public], 

33) 

34 

35BenefitGrantID = Annotated[UUID4, Path(description="The benefit grant ID.")] 

36BenefitGrantNotFound = { 

37 "description": "Benefit grant not found.", 

38 "model": ResourceNotFound.schema(), 

39} 

40 

41ListSorting = Annotated[ 

42 list[Sorting[CustomerBenefitGrantSortProperty]], 

43 Depends( 

44 SortingGetter( 

45 CustomerBenefitGrantSortProperty, ["product_benefit", "-granted_at"] 

46 ) 

47 ), 

48] 

49 

50 

51@router.get( 

52 "/", 

53 summary="List Benefit Grants", 

54 response_model=ListResource[CustomerBenefitGrant], 

55) 

56async def list( 

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 ) 

89 

90 return ListResource.from_paginated_results( 

91 [CustomerBenefitGrantAdapter.validate_python(result) for result in results], 

92 count, 

93 pagination, 

94 ) 

95 

96 

97@router.get( 

98 "/{id}", 

99 summary="Get Benefit Grant", 

100 response_model=CustomerBenefitGrant, 

101 responses={404: BenefitGrantNotFound}, 

102) 

103async def get( 

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 ) 

112 

113 if benefit_grant is None: 113 ↛ 116line 113 didn't jump to line 116 because the condition on line 113 was always true

114 raise ResourceNotFound() 

115 

116 return benefit_grant 

117 

118 

119@router.patch( 

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( 

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 ) 

142 

143 if benefit_grant is None: 143 ↛ 146line 143 didn't jump to line 146 because the condition on line 143 was always true

144 raise ResourceNotFound() 

145 

146 return await customer_benefit_grant_service.update( 

147 session, benefit_grant, benefit_grant_update 

148 )