Coverage for polar/benefit/strategies/downloadables/service.py: 58%

1from __future__ import annotations 

2 

3from typing import Any, cast 

4from uuid import UUID 

5 

6import structlog 

7 

8from polar.auth.models import AuthSubject 

9from polar.customer_portal.service.downloadables import ( 

10 downloadable as downloadable_service, 

11) 

12from polar.logging import Logger 

13from polar.models import Benefit, Customer, Organization, User 

14 

15from ..base.service import BenefitServiceProtocol 

16from . import schemas 

17from .properties import ( 

18 BenefitDownloadablesProperties, 

19 BenefitGrantDownloadablesProperties, 

20) 

21 

22log: Logger = structlog.get_logger() 

23 

24 

25def get_active_file_ids(properties: BenefitDownloadablesProperties) -> list[UUID]: 

26 schema = schemas.BenefitDownloadablesProperties(**properties) 

27 return schemas.get_active_file_ids(schema) 

28 

29 

30class BenefitDownloadablesService( 

31 BenefitServiceProtocol[ 

32 BenefitDownloadablesProperties, BenefitGrantDownloadablesProperties 

33 ] 

34): 

35 async def grant( 

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 {} 

48 

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)) 

59 

60 return { 

61 "files": granted, 

62 } 

63 

64 async def cycle( 

65 self, 

66 benefit: Benefit, 

67 customer: Customer, 

68 grant_properties: BenefitGrantDownloadablesProperties, 

69 *, 

70 attempt: int = 1, 

71 ) -> BenefitGrantDownloadablesProperties: 

72 return grant_properties 

73 

74 async def revoke( 

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 {} 

88 

89 async def requires_update( 

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 

96 

97 async def validate_properties( 

98 self, auth_subject: AuthSubject[User | Organization], properties: dict[str, Any] 

99 ) -> BenefitDownloadablesProperties: 

100 return cast(BenefitDownloadablesProperties, properties)