Coverage for polar/benefit/strategies/downloadables/service.py: 42%
39 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.customer_portal.service.downloadables import ( 1a
10 downloadable as downloadable_service,
11)
12from polar.logging import Logger 1a
13from polar.models import Benefit, Customer, Organization, User 1a
15from ..base.service import BenefitServiceProtocol 1a
16from . import schemas 1a
17from .properties import ( 1a
18 BenefitDownloadablesProperties,
19 BenefitGrantDownloadablesProperties,
20)
22log: Logger = structlog.get_logger() 1a
25def get_active_file_ids(properties: BenefitDownloadablesProperties) -> list[UUID]: 1a
26 schema = schemas.BenefitDownloadablesProperties(**properties)
27 return schemas.get_active_file_ids(schema)
30class BenefitDownloadablesService( 1a
31 BenefitServiceProtocol[
32 BenefitDownloadablesProperties, BenefitGrantDownloadablesProperties
33 ]
34):
35 async def grant( 1a
36 self,
37 benefit: Benefit,
38 customer: Customer,
39 grant_properties: BenefitGrantDownloadablesProperties,
40 *,
41 update: bool = False,
42 attempt: int = 1,
43 ) -> BenefitGrantDownloadablesProperties:
44 properties = self._get_properties(benefit)
45 file_ids = get_active_file_ids(properties)
46 if not file_ids:
47 return {}
49 granted = []
50 for file_id in file_ids:
51 downloadable = await downloadable_service.grant_for_benefit_file(
52 self.session,
53 customer=customer,
54 benefit_id=benefit.id,
55 file_id=file_id,
56 )
57 if downloadable:
58 granted.append(str(downloadable.file_id))
60 return {
61 "files": granted,
62 }
64 async def cycle( 1a
65 self,
66 benefit: Benefit,
67 customer: Customer,
68 grant_properties: BenefitGrantDownloadablesProperties,
69 *,
70 attempt: int = 1,
71 ) -> BenefitGrantDownloadablesProperties:
72 return grant_properties
74 async def revoke( 1a
75 self,
76 benefit: Benefit,
77 customer: Customer,
78 grant_properties: BenefitGrantDownloadablesProperties,
79 *,
80 attempt: int = 1,
81 ) -> BenefitGrantDownloadablesProperties:
82 await downloadable_service.revoke_for_benefit(
83 self.session,
84 customer=customer,
85 benefit_id=benefit.id,
86 )
87 return {}
89 async def requires_update( 1a
90 self, benefit: Benefit, previous_properties: BenefitDownloadablesProperties
91 ) -> bool:
92 properties = self._get_properties(benefit)
93 new_file_ids = set(get_active_file_ids(properties))
94 previous_file_ids = set(get_active_file_ids(previous_properties))
95 return new_file_ids != previous_file_ids
97 async def validate_properties( 1a
98 self, auth_subject: AuthSubject[User | Organization], properties: dict[str, Any]
99 ) -> BenefitDownloadablesProperties:
100 return cast(BenefitDownloadablesProperties, properties)