Coverage for polar/benefit/repository.py: 32%

27 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-12-05 16:17 +0000

1from uuid import UUID 1a

2 

3from sqlalchemy import Select, select 1a

4from sqlalchemy.orm import joinedload 1a

5 

6from polar.auth.models import AuthSubject, Organization, User, is_organization, is_user 1a

7from polar.kit.repository import ( 1a

8 Options, 

9 RepositoryBase, 

10 RepositorySoftDeletionIDMixin, 

11 RepositorySoftDeletionMixin, 

12 RepositorySortingMixin, 

13 SortingClause, 

14) 

15from polar.models import Benefit, UserOrganization 1a

16 

17from .sorting import BenefitSortProperty 1a

18 

19 

20class BenefitRepository( 1a

21 RepositorySortingMixin[Benefit, BenefitSortProperty], 

22 RepositorySoftDeletionIDMixin[Benefit, UUID], 

23 RepositorySoftDeletionMixin[Benefit], 

24 RepositoryBase[Benefit], 

25): 

26 model = Benefit 1a

27 

28 def get_eager_options(self) -> Options: 1a

29 return (joinedload(Benefit.organization),) 

30 

31 def get_readable_statement( 1a

32 self, auth_subject: AuthSubject[User | Organization] 

33 ) -> Select[tuple[Benefit]]: 

34 statement = self.get_base_statement() 

35 

36 if is_user(auth_subject): 

37 user = auth_subject.subject 

38 statement = statement.where( 

39 Benefit.organization_id.in_( 

40 select(UserOrganization.organization_id).where( 

41 UserOrganization.user_id == user.id, 

42 UserOrganization.deleted_at.is_(None), 

43 ) 

44 ) 

45 ) 

46 elif is_organization(auth_subject): 

47 statement = statement.where( 

48 Benefit.organization_id == auth_subject.subject.id, 

49 ) 

50 

51 return statement 

52 

53 def get_sorting_clause(self, property: BenefitSortProperty) -> SortingClause: 1a

54 match property: 

55 case BenefitSortProperty.created_at: 

56 return Benefit.created_at 

57 case BenefitSortProperty.description: 

58 return Benefit.description 

59 case BenefitSortProperty.type: 

60 return Benefit.type