Coverage for polar/trial_redemption/service.py: 56%

1from email_validator.validate_email import validate_email 1 ctx1a

2 

3from polar.models import Customer, Organization, Product, TrialRedemption 1 ctx1a

4from polar.postgres import AsyncSession 1 ctx1a

5from polar.trial_redemption.repository import TrialRedemptionRepository 1 ctx1a

6 

7 

8class TrialRedemptionService: 1 ctx1a

9 async def check_trial_already_redeemed( 1 ctx1a

10 self, 

11 session: AsyncSession, 

12 organization: Organization, 

13 *, 

14 customer: Customer, 

15 product: Product | None = None, 

16 payment_method_fingerprint: str | None = None, 

17 ) -> bool: 

18 repository = TrialRedemptionRepository.from_session(session) 

19 trial_redemptions = await repository.get_all_by_organization_and_hints( 

20 organization.id, 

21 customer_email=self._get_unaliased_email(customer.email), 

22 product=product.id if product else None, 

23 payment_method_fingerprint=payment_method_fingerprint, 

24 ) 

25 return len(trial_redemptions) > 0 

26 

27 async def create_trial_redemption( 1 ctx1a

28 self, 

29 session: AsyncSession, 

30 *, 

31 customer: Customer, 

32 product: Product | None = None, 

33 payment_method_fingerprint: str | None = None, 

34 ) -> TrialRedemption: 

35 repository = TrialRedemptionRepository.from_session(session) 

36 return await repository.create( 

37 TrialRedemption( 

38 customer_email=self._get_unaliased_email(customer.email), 

39 customer=customer, 

40 product=product, 

41 payment_method_fingerprint=payment_method_fingerprint, 

42 ) 

43 ) 

44 

45 def _get_unaliased_email(self, email: str) -> str: 1 ctx1a

46 parsed_email = validate_email(email, check_deliverability=False) 

47 return f"{parsed_email.local_part.split('+', 1)[0]}@{parsed_email.domain}" 

48 

49 

50trial_redemption = TrialRedemptionService() 1 ctx1a