Coverage for polar/license_key/endpoints.py: 42%
62 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 15:52 +0000
1from fastapi import Depends, Query 1a
2from pydantic import UUID4 1a
4from polar.benefit.schemas import BenefitID 1a
5from polar.exceptions import ResourceNotFound 1a
6from polar.kit.db.postgres import AsyncReadSession, AsyncSession 1a
7from polar.kit.pagination import ListResource, PaginationParamsQuery 1a
8from polar.kit.schemas import MultipleQueryFilter 1a
9from polar.models import LicenseKey, LicenseKeyActivation 1a
10from polar.openapi import APITag 1a
11from polar.organization.schemas import OrganizationID 1a
12from polar.postgres import get_db_read_session, get_db_session 1a
13from polar.routing import APIRouter 1a
15from . import auth 1a
16from .repository import LicenseKeyRepository 1a
17from .schemas import ( 1a
18 ActivationNotPermitted,
19 LicenseKeyActivate,
20 LicenseKeyActivationRead,
21 LicenseKeyDeactivate,
22 LicenseKeyRead,
23 LicenseKeyUpdate,
24 LicenseKeyValidate,
25 LicenseKeyWithActivations,
26 NotFoundResponse,
27 UnauthorizedResponse,
28 ValidatedLicenseKey,
29)
30from .service import license_key as license_key_service 1a
32router = APIRouter(prefix="/license-keys", tags=["license_keys", APITag.public]) 1a
35@router.get( 1a
36 "/",
37 summary="List License Keys",
38 response_model=ListResource[LicenseKeyRead],
39 responses={
40 401: UnauthorizedResponse,
41 404: NotFoundResponse,
42 },
43)
44async def list( 1a
45 auth_subject: auth.LicenseKeysRead,
46 pagination: PaginationParamsQuery,
47 organization_id: MultipleQueryFilter[OrganizationID] | None = Query(
48 None, title="OrganizationID Filter", description="Filter by organization ID."
49 ),
50 benefit_id: MultipleQueryFilter[BenefitID] | None = Query(
51 None, title="BenefitID Filter", description="Filter by benefit ID."
52 ),
53 session: AsyncReadSession = Depends(get_db_read_session),
54) -> ListResource[LicenseKeyRead]:
55 """Get license keys connected to the given organization & filters."""
56 results, count = await license_key_service.list(
57 session,
58 auth_subject,
59 organization_id=organization_id,
60 benefit_id=benefit_id,
61 pagination=pagination,
62 )
64 return ListResource.from_paginated_results(
65 [LicenseKeyRead.model_validate(result) for result in results],
66 count,
67 pagination,
68 )
71@router.get( 1a
72 "/{id}",
73 summary="Get License Key",
74 response_model=LicenseKeyWithActivations,
75 responses={
76 401: UnauthorizedResponse,
77 404: NotFoundResponse,
78 },
79)
80async def get( 1a
81 auth_subject: auth.LicenseKeysRead,
82 id: UUID4,
83 session: AsyncReadSession = Depends(get_db_read_session),
84) -> LicenseKey:
85 """Get a license key."""
86 lk = await license_key_service.get(session, auth_subject, id)
87 if not lk:
88 raise ResourceNotFound()
90 return lk
93@router.patch( 1a
94 "/{id}",
95 summary="Update License Key",
96 response_model=LicenseKeyRead,
97 responses={
98 401: UnauthorizedResponse,
99 404: NotFoundResponse,
100 },
101)
102async def update( 1a
103 auth_subject: auth.LicenseKeysWrite,
104 id: UUID4,
105 updates: LicenseKeyUpdate,
106 session: AsyncSession = Depends(get_db_session),
107) -> LicenseKey:
108 """Update a license key."""
109 lk = await license_key_service.get(session, auth_subject, id)
110 if not lk:
111 raise ResourceNotFound()
113 updated = await license_key_service.update(session, license_key=lk, updates=updates)
114 return updated
117@router.get( 1a
118 "/{id}/activations/{activation_id}",
119 summary="Get Activation",
120 response_model=LicenseKeyActivationRead,
121 responses={
122 401: UnauthorizedResponse,
123 404: NotFoundResponse,
124 },
125)
126async def get_activation( 1a
127 auth_subject: auth.LicenseKeysRead,
128 id: UUID4,
129 activation_id: UUID4,
130 session: AsyncReadSession = Depends(get_db_read_session),
131) -> LicenseKeyActivation:
132 """Get a license key activation."""
133 lk = await license_key_service.get(session, auth_subject, id)
134 if not lk:
135 raise ResourceNotFound()
137 activation = await license_key_service.get_activation_or_raise(
138 session,
139 license_key=lk,
140 activation_id=activation_id,
141 )
142 return activation
145@router.post( 1a
146 "/validate",
147 summary="Validate License Key",
148 response_model=ValidatedLicenseKey,
149 responses={
150 404: NotFoundResponse,
151 },
152)
153async def validate( 1a
154 auth_subject: auth.LicenseKeysWrite,
155 validate: LicenseKeyValidate,
156 session: AsyncSession = Depends(get_db_session),
157) -> LicenseKey:
158 """Validate a license key."""
159 repository = LicenseKeyRepository.from_session(session)
160 license_key = await repository.get_readable_by_key(
161 validate.key,
162 validate.organization_id,
163 auth_subject,
164 options=repository.get_eager_options(),
165 )
167 if license_key is None:
168 raise ResourceNotFound()
170 return await license_key_service.validate(
171 session, license_key=license_key, validate=validate
172 )
175@router.post( 1a
176 "/activate",
177 summary="Activate License Key",
178 response_model=LicenseKeyActivationRead,
179 responses={
180 403: ActivationNotPermitted,
181 404: NotFoundResponse,
182 },
183)
184async def activate( 1a
185 auth_subject: auth.LicenseKeysWrite,
186 activate: LicenseKeyActivate,
187 session: AsyncSession = Depends(get_db_session),
188) -> LicenseKeyActivation:
189 """Activate a license key instance."""
190 repository = LicenseKeyRepository.from_session(session)
191 license_key = await repository.get_readable_by_key(
192 activate.key,
193 activate.organization_id,
194 auth_subject,
195 options=repository.get_eager_options(),
196 )
198 if license_key is None:
199 raise ResourceNotFound()
201 return await license_key_service.activate(
202 session, license_key=license_key, activate=activate
203 )
206@router.post( 1a
207 "/deactivate",
208 summary="Deactivate License Key",
209 status_code=204,
210 responses={
211 204: {"description": "License key activation deactivated."},
212 404: NotFoundResponse,
213 },
214)
215async def deactivate( 1a
216 auth_subject: auth.LicenseKeysWrite,
217 deactivate: LicenseKeyDeactivate,
218 session: AsyncSession = Depends(get_db_session),
219) -> None:
220 """Deactivate a license key instance."""
221 repository = LicenseKeyRepository.from_session(session)
222 license_key = await repository.get_readable_by_key(
223 deactivate.key,
224 deactivate.organization_id,
225 auth_subject,
226 options=repository.get_eager_options(),
227 )
229 if license_key is None:
230 raise ResourceNotFound()
232 await license_key_service.deactivate(
233 session, license_key=license_key, deactivate=deactivate
234 )