Coverage for polar/customer_portal/endpoints/license_keys.py: 55%
43 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 17:15 +0000
1from typing import cast 1a
3from fastapi import Depends, Query 1a
4from pydantic import UUID4 1a
6from polar.benefit.schemas import BenefitID 1a
7from polar.benefit.strategies.license_keys.properties import ( 1a
8 BenefitLicenseKeysProperties,
9)
10from polar.exceptions import ResourceNotFound 1a
11from polar.kit.db.postgres import AsyncSession 1a
12from polar.kit.pagination import ListResource, PaginationParamsQuery 1a
13from polar.license_key.schemas import ( 1a
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 1a
26from polar.models import LicenseKey, LicenseKeyActivation 1a
27from polar.openapi import APITag 1a
28from polar.postgres import get_db_session 1a
29from polar.routing import APIRouter 1a
31from .. import auth 1a
33router = APIRouter(prefix="/license-keys", tags=["license_keys", APITag.public]) 1a
36@router.get( 1a
37 "/",
38 summary="List License Keys",
39 response_model=ListResource[LicenseKeyRead],
40 responses={
41 401: UnauthorizedResponse,
42 404: NotFoundResponse,
43 },
44)
45async def list( 1a
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 )
60 return ListResource.from_paginated_results(
61 [LicenseKeyRead.model_validate(result) for result in results],
62 count,
63 pagination,
64 )
67@router.get( 1a
68 "/{id}",
69 summary="Get License Key",
70 response_model=LicenseKeyWithActivations,
71 responses={404: NotFoundResponse},
72)
73async def get( 1a
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:
81 raise ResourceNotFound()
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 = []
89 return ret
92@router.post( 1a
93 "/validate",
94 summary="Validate License Key",
95 response_model=ValidatedLicenseKey,
96 responses={
97 404: NotFoundResponse,
98 },
99)
100async def validate( 1ab
101 validate: LicenseKeyValidate,
102 session: AsyncSession = Depends(get_db_session),
103) -> LicenseKey:
104 """
105 Validate a license key.
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 )
122@router.post( 1a
123 "/activate",
124 summary="Activate License Key",
125 response_model=LicenseKeyActivationRead,
126 responses={
127 403: ActivationNotPermitted,
128 404: NotFoundResponse,
129 },
130)
131async def activate( 1ab
132 activate: LicenseKeyActivate,
133 session: AsyncSession = Depends(get_db_session),
134) -> LicenseKeyActivation:
135 """
136 Activate a license key instance.
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 )
153@router.post( 1a
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( 1ab
163 deactivate: LicenseKeyDeactivate,
164 session: AsyncSession = Depends(get_db_session),
165) -> None:
166 """
167 Deactivate a license key instance.
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)