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

21 statements  

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

1import uuid 1a

2from collections.abc import Sequence 1a

3 

4from polar.auth.models import AuthSubject 1a

5from polar.kit.db.postgres import AsyncSession 1a

6from polar.kit.pagination import PaginationParams 1a

7from polar.kit.services import ResourceServiceReader 1a

8from polar.kit.sorting import Sorting 1a

9from polar.models import Customer, Wallet 1a

10 

11from ..repository.wallet import CustomerWalletRepository 1a

12from ..sorting.wallet import CustomerWalletSortProperty 1a

13 

14 

15class CustomerWalletService(ResourceServiceReader[Wallet]): 1a

16 async def list( 1a

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( 1a

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) 1a