Coverage for polar/discount/endpoints.py: 88%

1from typing import Annotated 

2 

3from fastapi import Body, Depends, Path, Query 

4from pydantic import UUID4 

5 

6from polar.exceptions import ResourceNotFound 

7from polar.kit.pagination import ListResource, PaginationParamsQuery 

8from polar.kit.schemas import MultipleQueryFilter, SetSchemaReference 

9from polar.models import Discount 

10from polar.openapi import APITag 

11from polar.organization.schemas import OrganizationID 

12from polar.postgres import AsyncSession, get_db_session 

13from polar.routing import APIRouter 

14 

15from . import auth, sorting 

16from .schemas import Discount as DiscountSchema 

17from .schemas import ( 

18 DiscountAdapter, 

19 DiscountCreate, 

20 DiscountUpdate, 

21) 

22from .service import discount as discount_service 

23 

24router = APIRouter(prefix="/discounts", tags=["discounts", APITag.public]) 

25 

26 

27DiscountID = Annotated[UUID4, Path(description="The discount ID.")] 

28DiscountNotFound = { 

29 "description": "Discount not found.", 

30 "model": ResourceNotFound.schema(), 

31} 

32 

33 

34@router.get("/", summary="List Discounts", response_model=ListResource[DiscountSchema]) 

35async def list( 

36 auth_subject: auth.DiscountRead, 

37 pagination: PaginationParamsQuery, 

38 sorting: sorting.ListSorting, 

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

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

41 ), 

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

43 session: AsyncSession = Depends(get_db_session), 

44) -> ListResource[DiscountSchema]: 

45 """List discounts.""" 

46 results, count = await discount_service.list( 

47 session, 

48 auth_subject, 

49 organization_id=organization_id, 

50 query=query, 

51 pagination=pagination, 

52 sorting=sorting, 

53 ) 

54 

55 return ListResource.from_paginated_results( 

56 [DiscountAdapter.validate_python(result) for result in results], 

57 count, 

58 pagination, 

59 ) 

60 

61 

62@router.get( 

63 "/{id}", 

64 summary="Get Discount", 

65 response_model=DiscountSchema, 

66 responses={404: DiscountNotFound}, 

67) 

68async def get( 

69 id: DiscountID, 

70 auth_subject: auth.DiscountRead, 

71 session: AsyncSession = Depends(get_db_session), 

72) -> Discount: 

73 """Get a discount by ID.""" 

74 discount = await discount_service.get_by_id(session, auth_subject, id) 

75 

76 if discount is None: 76 ↛ 79line 76 didn't jump to line 79 because the condition on line 76 was always true

77 raise ResourceNotFound() 

78 

79 return discount 

80 

81 

82@router.post( 

83 "/", 

84 response_model=DiscountSchema, 

85 status_code=201, 

86 summary="Create Discount", 

87 responses={201: {"description": "Discount created."}}, 

88) 

89async def create( 

90 auth_subject: auth.DiscountWrite, 

91 # This is a workaround for FastAPI bug: https://github.com/fastapi/fastapi/discussions/12941 

92 discount_create: Annotated[ 

93 DiscountCreate, 

94 Body(...), 

95 SetSchemaReference("DiscountCreate"), 

96 ], 

97 session: AsyncSession = Depends(get_db_session), 

98) -> Discount: 

99 """Create a discount.""" 

100 return await discount_service.create(session, discount_create, auth_subject) 

101 

102 

103@router.patch( 

104 "/{id}", 

105 response_model=DiscountSchema, 

106 summary="Update Discount", 

107 responses={ 

108 200: {"description": "Discount updated."}, 

109 404: DiscountNotFound, 

110 }, 

111) 

112async def update( 

113 id: DiscountID, 

114 discount_update: DiscountUpdate, 

115 auth_subject: auth.DiscountWrite, 

116 session: AsyncSession = Depends(get_db_session), 

117) -> Discount: 

118 """Update a discount.""" 

119 discount = await discount_service.get_by_id(session, auth_subject, id) 

120 

121 if discount is None: 121 ↛ 124line 121 didn't jump to line 124 because the condition on line 121 was always true

122 raise ResourceNotFound() 

123 

124 return await discount_service.update(session, discount, discount_update) 

125 

126 

127@router.delete( 

128 "/{id}", 

129 status_code=204, 

130 summary="Delete Discount", 

131 responses={ 

132 204: {"description": "Discount deleted."}, 

133 404: DiscountNotFound, 

134 }, 

135) 

136async def delete( 

137 id: DiscountID, 

138 auth_subject: auth.DiscountWrite, 

139 session: AsyncSession = Depends(get_db_session), 

140) -> None: 

141 """Delete a discount.""" 

142 discount = await discount_service.get_by_id(session, auth_subject, id) 

143 

144 if discount is None: 144 ↛ 147line 144 didn't jump to line 147 because the condition on line 144 was always true

145 raise ResourceNotFound() 

146 

147 await discount_service.delete(session, discount)