Coverage for polar/benefit/grant/scope.py: 26%

1from polar.exceptions import PolarError 1 ctx1a

2from polar.models.benefit_grant import BenefitGrantScope, BenefitGrantScopeArgs 1 ctx1a

3from polar.order.repository import OrderRepository 1 ctx1a

4from polar.postgres import AsyncSession 1 ctx1a

5from polar.subscription.repository import SubscriptionRepository 1 ctx1a

6 

7 

8class BenefitGrantScopeError(PolarError): ... 1 ctx1a

9 

10 

11class InvalidScopeError(BenefitGrantScopeError): 1 ctx1a

12 def __init__(self, scope: BenefitGrantScopeArgs) -> None: 1 ctx1a

13 self.scope = scope 

14 message = "The provided scope is invalid." 

15 super().__init__(message, 500) 

16 

17 

18async def resolve_scope( 1 ctx1a

19 session: AsyncSession, scope: BenefitGrantScopeArgs 

20) -> BenefitGrantScope: 

21 resolved_scope: BenefitGrantScope = {} 

22 if subscription_id := scope.get("subscription_id"): 

23 subscription_repository = SubscriptionRepository.from_session(session) 

24 subscription = await subscription_repository.get_by_id(subscription_id) 

25 if subscription is None: 

26 raise InvalidScopeError(scope) 

27 resolved_scope["subscription"] = subscription 

28 if order_id := scope.get("order_id"): 

29 order_repository = OrderRepository.from_session(session) 

30 order = await order_repository.get_by_id(order_id) 

31 if order is None: 

32 raise InvalidScopeError(scope) 

33 resolved_scope["order"] = order 

34 return resolved_scope 

35 

36 

37def scope_to_args(scope: BenefitGrantScope) -> BenefitGrantScopeArgs: 1 ctx1a

38 args: BenefitGrantScopeArgs = {} 

39 if subscription := scope.get("subscription"): 

40 args["subscription_id"] = subscription.id 

41 if order := scope.get("order"): 

42 args["order_id"] = order.id 

43 return args