Coverage for polar/customer_portal/endpoints/license_keys.py: 74%

1from typing import cast 

2 

3from fastapi import Depends, Query 

4from pydantic import UUID4 

5 

6from polar.benefit.schemas import BenefitID 

7from polar.benefit.strategies.license_keys.properties import ( 

8 BenefitLicenseKeysProperties, 

9) 

10from polar.exceptions import ResourceNotFound 

11from polar.kit.db.postgres import AsyncSession 

12from polar.kit.pagination import ListResource, PaginationParamsQuery 

13from polar.license_key.schemas import ( 

14 ActivationNotPermitted, 

15 LicenseKeyActivate, 

16 LicenseKeyActivationRead, 

17 LicenseKeyDeactivate, 

18 LicenseKeyRead, 

19 LicenseKeyValidate, 

20 LicenseKeyWithActivations, 

21 NotFoundResponse, 

22 UnauthorizedResponse, 

23 ValidatedLicenseKey, 

24) 

25from polar.license_key.service import license_key as license_key_service 

26from polar.models import LicenseKey, LicenseKeyActivation 

27from polar.openapi import APITag 

28from polar.postgres import get_db_session 

29from polar.routing import APIRouter 

30 

31from .. import auth 

32 

33router = APIRouter(prefix="/license-keys", tags=["license_keys", APITag.public]) 

34 

35 

36@router.get( 

37 "/", 

38 summary="List License Keys", 

39 response_model=ListResource[LicenseKeyRead], 

40 responses={ 

41 401: UnauthorizedResponse, 

42 404: NotFoundResponse, 

43 }, 

44) 

45async def list( 

46 auth_subject: auth.CustomerPortalRead, 

47 pagination: PaginationParamsQuery, 

48 benefit_id: BenefitID | None = Query( 

49 None, description="Filter by a specific benefit" 

50 ), 

51 session: AsyncSession = Depends(get_db_session), 

52) -> ListResource[LicenseKeyRead]: 

53 results, count = await license_key_service.get_customer_list( 

54 session, 

55 auth_subject, 

56 benefit_id=benefit_id, 

57 pagination=pagination, 

58 ) 

59 

60 return ListResource.from_paginated_results( 

61 [LicenseKeyRead.model_validate(result) for result in results], 

62 count, 

63 pagination, 

64 ) 

65 

66 

67@router.get( 

68 "/{id}", 

69 summary="Get License Key", 

70 response_model=LicenseKeyWithActivations, 

71 responses={404: NotFoundResponse}, 

72) 

73async def get( 

74 auth_subject: auth.CustomerPortalRead, 

75 id: UUID4, 

76 session: AsyncSession = Depends(get_db_session), 

77) -> LicenseKeyWithActivations: 

78 """Get a license key.""" 

79 lk = await license_key_service.get_customer_license_key(session, auth_subject, id) 

80 if not lk: 80 ↛ 83line 80 didn't jump to line 83 because the condition on line 80 was always true

81 raise ResourceNotFound() 

82 

83 ret = LicenseKeyWithActivations.model_validate(lk) 

84 properties = cast(BenefitLicenseKeysProperties, lk.benefit.properties) 

85 activations = properties.get("activations") 

86 if not (activations and activations.get("enable_customer_admin")): 

87 ret.activations = [] 

88 

89 return ret 

90 

91 

92@router.post( 

93 "/validate", 

94 summary="Validate License Key", 

95 response_model=ValidatedLicenseKey, 

96 responses={ 

97 404: NotFoundResponse, 

98 }, 

99) 

100async def validate( 

101 validate: LicenseKeyValidate, 

102 session: AsyncSession = Depends(get_db_session), 

103) -> LicenseKey: 

104 """ 

105 Validate a license key. 

106 

107 > This endpoint doesn't require authentication and can be safely used on a public 

108 > client, like a desktop application or a mobile app. 

109 > If you plan to validate a license key on a server, use the `/v1/license-keys/validate` 

110 > endpoint instead. 

111 """ 

112 license_key = await license_key_service.get_or_raise_by_key( 

113 session, 

114 organization_id=validate.organization_id, 

115 key=validate.key, 

116 ) 

117 return await license_key_service.validate( 

118 session, license_key=license_key, validate=validate 

119 ) 

120 

121 

122@router.post( 

123 "/activate", 

124 summary="Activate License Key", 

125 response_model=LicenseKeyActivationRead, 

126 responses={ 

127 403: ActivationNotPermitted, 

128 404: NotFoundResponse, 

129 }, 

130) 

131async def activate( 

132 activate: LicenseKeyActivate, 

133 session: AsyncSession = Depends(get_db_session), 

134) -> LicenseKeyActivation: 

135 """ 

136 Activate a license key instance. 

137 

138 > This endpoint doesn't require authentication and can be safely used on a public 

139 > client, like a desktop application or a mobile app. 

140 > If you plan to validate a license key on a server, use the `/v1/license-keys/activate` 

141 > endpoint instead. 

142 """ 

143 lk = await license_key_service.get_or_raise_by_key( 

144 session, 

145 organization_id=activate.organization_id, 

146 key=activate.key, 

147 ) 

148 return await license_key_service.activate( 

149 session, license_key=lk, activate=activate 

150 ) 

151 

152 

153@router.post( 

154 "/deactivate", 

155 summary="Deactivate License Key", 

156 status_code=204, 

157 responses={ 

158 204: {"description": "License key activation deactivated."}, 

159 404: NotFoundResponse, 

160 }, 

161) 

162async def deactivate( 

163 deactivate: LicenseKeyDeactivate, 

164 session: AsyncSession = Depends(get_db_session), 

165) -> None: 

166 """ 

167 Deactivate a license key instance. 

168 

169 > This endpoint doesn't require authentication and can be safely used on a public 

170 > client, like a desktop application or a mobile app. 

171 > If you plan to validate a license key on a server, use the `/v1/license-keys/deactivate` 

172 > endpoint instead. 

173 """ 

174 lk = await license_key_service.get_or_raise_by_key( 

175 session, 

176 organization_id=deactivate.organization_id, 

177 key=deactivate.key, 

178 ) 

179 await license_key_service.deactivate(session, license_key=lk, deactivate=deactivate)