Coverage for polar/customer_portal/service/customer_meter.py: 50%

1import uuid 1 ctx1a

2from collections.abc import Sequence 1 ctx1a

3 

4from sqlalchemy.orm import joinedload 1 ctx1a

5from sqlalchemy.orm.strategy_options import contains_eager 1 ctx1a

6 

7from polar.auth.models import AuthSubject 1 ctx1a

8from polar.kit.db.postgres import AsyncSession 1 ctx1a

9from polar.kit.pagination import PaginationParams 1 ctx1a

10from polar.kit.sorting import Sorting 1 ctx1a

11from polar.models import Customer, CustomerMeter, Meter 1 ctx1a

12 

13from ..repository.customer_meter import CustomerMeterRepository 1 ctx1a

14from ..sorting.customer_meter import CustomerCustomerMeterSortProperty 1 ctx1a

15 

16 

17class CustomerMeterService: 1 ctx1a

18 async def list( 1 ctx1a

19 self, 

20 session: AsyncSession, 

21 auth_subject: AuthSubject[Customer], 

22 *, 

23 meter_id: Sequence[uuid.UUID] | None = None, 

24 query: str | None = None, 

25 pagination: PaginationParams, 

26 sorting: list[Sorting[CustomerCustomerMeterSortProperty]] = [ 

27 (CustomerCustomerMeterSortProperty.modified_at, True) 

28 ], 

29 ) -> tuple[Sequence[CustomerMeter], int]: 

30 repository = CustomerMeterRepository.from_session(session) 

31 statement = ( 

32 repository.get_readable_statement(auth_subject) 

33 .join(CustomerMeter.meter) 

34 .options( 

35 contains_eager(CustomerMeter.meter), 

36 ) 

37 ) 

38 

39 if meter_id is not None: 

40 statement = statement.where(CustomerMeter.meter_id.in_(meter_id)) 

41 

42 if query is not None: 

43 statement = statement.where(Meter.name.ilike(f"%{query}%")) 

44 

45 statement = repository.apply_sorting(statement, sorting) 

46 

47 return await repository.paginate( 

48 statement, limit=pagination.limit, page=pagination.page 

49 ) 

50 

51 async def get( 1 ctx1a

52 self, 

53 session: AsyncSession, 

54 auth_subject: AuthSubject[Customer], 

55 id: uuid.UUID, 

56 ) -> CustomerMeter | None: 

57 repository = CustomerMeterRepository.from_session(session) 

58 statement = ( 

59 repository.get_readable_statement(auth_subject) 

60 .where(CustomerMeter.id == id) 

61 .options(joinedload(CustomerMeter.meter)) 

62 ) 

63 return await repository.get_one_or_none(statement) 

64 

65 

66customer_meter = CustomerMeterService() 1 ctx1a