Coverage for polar/benefit/strategies/license_keys/service.py: 44%
37 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 17:15 +0000
1from __future__ import annotations 1a
3from typing import Any, cast 1a
4from uuid import UUID 1a
6import structlog 1a
8from polar.auth.models import AuthSubject 1a
9from polar.license_key.service import license_key as license_key_service 1a
10from polar.logging import Logger 1a
11from polar.models import Benefit, Customer, Organization, User 1a
13from ..base.service import BenefitServiceProtocol 1a
14from .properties import BenefitGrantLicenseKeysProperties, BenefitLicenseKeysProperties 1a
16log: Logger = structlog.get_logger() 1a
19class BenefitLicenseKeysService( 1a
20 BenefitServiceProtocol[
21 BenefitLicenseKeysProperties, BenefitGrantLicenseKeysProperties
22 ]
23):
24 should_revoke_individually = True 1a
26 async def grant( 1a
27 self,
28 benefit: Benefit,
29 customer: Customer,
30 grant_properties: BenefitGrantLicenseKeysProperties,
31 *,
32 update: bool = False,
33 attempt: int = 1,
34 ) -> BenefitGrantLicenseKeysProperties:
35 current_lk_id = None
36 if update and "license_key_id" in grant_properties:
37 current_lk_id = UUID(grant_properties["license_key_id"])
39 key = await license_key_service.customer_grant(
40 self.session,
41 customer=customer,
42 benefit=benefit,
43 license_key_id=current_lk_id,
44 )
45 return {
46 "license_key_id": str(key.id),
47 "display_key": key.display_key,
48 }
50 async def cycle( 1a
51 self,
52 benefit: Benefit,
53 customer: Customer,
54 grant_properties: BenefitGrantLicenseKeysProperties,
55 *,
56 attempt: int = 1,
57 ) -> BenefitGrantLicenseKeysProperties:
58 return grant_properties
60 async def revoke( 1a
61 self,
62 benefit: Benefit,
63 customer: Customer,
64 grant_properties: BenefitGrantLicenseKeysProperties,
65 *,
66 attempt: int = 1,
67 ) -> BenefitGrantLicenseKeysProperties:
68 license_key_id = grant_properties.get("license_key_id")
69 if not license_key_id:
70 log.info(
71 "license_key.revoke.skip",
72 customer_id=customer.id,
73 benefit_id=benefit.id,
74 message="No license key to revoke",
75 )
76 return grant_properties
78 await license_key_service.customer_revoke(
79 self.session,
80 customer=customer,
81 benefit=benefit,
82 license_key_id=UUID(license_key_id),
83 )
84 # Keep grant properties for reference
85 return grant_properties
87 async def requires_update( 1a
88 self,
89 benefit: Benefit,
90 previous_properties: BenefitLicenseKeysProperties,
91 ) -> bool:
92 c = self._get_properties(benefit)
93 pre = previous_properties
95 diff_expires = c.get("expires", None) != pre.get("expires", None)
96 diff_activations = c.get("activations", None) != pre.get("activations", None)
97 diff_usage = c.get("limit_usage", None) != pre.get("limit_usage", None)
98 return diff_expires or diff_activations or diff_usage
100 async def validate_properties( 1a
101 self, auth_subject: AuthSubject[User | Organization], properties: dict[str, Any]
102 ) -> BenefitLicenseKeysProperties:
103 return cast(BenefitLicenseKeysProperties, properties)