Coverage for polar/customer_portal/service/wallet.py: 100%

1import uuid 

2from collections.abc import Sequence 

3 

4from polar.auth.models import AuthSubject 

5from polar.kit.db.postgres import AsyncSession 

6from polar.kit.pagination import PaginationParams 

7from polar.kit.services import ResourceServiceReader 

8from polar.kit.sorting import Sorting 

9from polar.models import Customer, Wallet 

10 

11from ..repository.wallet import CustomerWalletRepository 

12from ..sorting.wallet import CustomerWalletSortProperty 

13 

14 

15class CustomerWalletService(ResourceServiceReader[Wallet]): 

16 async def list( 

17 self, 

18 session: AsyncSession, 

19 auth_subject: AuthSubject[Customer], 

20 *, 

21 pagination: PaginationParams, 

22 sorting: list[Sorting[CustomerWalletSortProperty]] = [ 

23 (CustomerWalletSortProperty.created_at, True) 

24 ], 

25 ) -> tuple[Sequence[Wallet], int]: 

26 repository = CustomerWalletRepository.from_session(session) 

27 statement = repository.get_readable_statement(auth_subject).options( 

28 *repository.get_eager_options() 

29 ) 

30 

31 statement = repository.apply_sorting(statement, sorting) 

32 

33 return await repository.paginate( 

34 statement, limit=pagination.limit, page=pagination.page 

35 ) 

36 

37 async def get_by_id( 

38 self, 

39 session: AsyncSession, 

40 auth_subject: AuthSubject[Customer], 

41 id: uuid.UUID, 

42 ) -> Wallet | None: 

43 repository = CustomerWalletRepository.from_session(session) 

44 statement = ( 

45 repository.get_readable_statement(auth_subject) 

46 .where(Wallet.id == id) 

47 .options(*repository.get_eager_options()) 

48 ) 

49 return await repository.get_one_or_none(statement) 

50 

51 

52customer_wallet = CustomerWalletService(Wallet)