Coverage for polar/trial_redemption/repository.py: 30%

1from collections.abc import Sequence 1 ctx1a

2from uuid import UUID 1 ctx1a

3 

4from sqlalchemy import ColumnExpressionArgument, func, or_, select 1 ctx1a

5 

6from polar.kit.repository import RepositoryBase 1 ctx1a

7from polar.models import Customer, TrialRedemption 1 ctx1a

8 

9 

10class TrialRedemptionRepository(RepositoryBase[TrialRedemption]): 1 ctx1a

11 model = TrialRedemption 1 ctx1a

12 

13 async def get_all_by_organization_and_hints( 1 ctx1a

14 self, 

15 organization: UUID, 

16 customer_email: str | None = None, 

17 payment_method_fingerprint: str | None = None, 

18 product: UUID | None = None, 

19 ) -> Sequence[TrialRedemption]: 

20 statement = ( 

21 select(TrialRedemption) 

22 .join(TrialRedemption.customer) 

23 .where(Customer.organization_id == organization) 

24 ) 

25 

26 if product is not None: 

27 statement = statement.where(TrialRedemption.product_id == product) 

28 

29 clauses: list[ColumnExpressionArgument[bool]] = [] 

30 

31 if customer_email is not None: 

32 clauses.append( 

33 func.lower(TrialRedemption.customer_email) == customer_email.lower() 

34 ) 

35 

36 if payment_method_fingerprint is not None: 

37 clauses.append( 

38 TrialRedemption.payment_method_fingerprint == payment_method_fingerprint 

39 ) 

40 

41 if clauses: 

42 statement = statement.where(or_(*clauses)) 

43 

44 return await self.get_all(statement)