Coverage for polar/customer_portal/endpoints/wallet.py: 93%

1from typing import Annotated 

2 

3from fastapi import Depends 

4 

5from polar.exceptions import ResourceNotFound 

6from polar.kit.db.postgres import AsyncSession 

7from polar.kit.pagination import ListResource, PaginationParamsQuery 

8from polar.kit.sorting import Sorting, SortingGetter 

9from polar.models import Wallet 

10from polar.openapi import APITag 

11from polar.postgres import get_db_session 

12from polar.routing import APIRouter 

13from polar.wallet.schemas import WalletID, WalletNotFound 

14 

15from .. import auth 

16from ..schemas.wallet import CustomerWallet 

17from ..service.wallet import customer_wallet as customer_wallet_service 

18from ..sorting.wallet import CustomerWalletSortProperty 

19 

20router = APIRouter(prefix="/wallets", tags=["wallets", APITag.public]) 

21 

22ListSorting = Annotated[ 

23 list[Sorting[CustomerWalletSortProperty]], 

24 Depends(SortingGetter(CustomerWalletSortProperty, ["-created_at"])), 

25] 

26 

27 

28@router.get("/", summary="List Wallets", response_model=ListResource[CustomerWallet]) 

29async def list( 

30 auth_subject: auth.CustomerPortalRead, 

31 pagination: PaginationParamsQuery, 

32 sorting: ListSorting, 

33 session: AsyncSession = Depends(get_db_session), 

34) -> ListResource[CustomerWallet]: 

35 """List wallets of the authenticated customer.""" 

36 results, count = await customer_wallet_service.list( 

37 session, 

38 auth_subject, 

39 pagination=pagination, 

40 sorting=sorting, 

41 ) 

42 

43 return ListResource.from_paginated_results( 

44 [CustomerWallet.model_validate(result) for result in results], 

45 count, 

46 pagination, 

47 ) 

48 

49 

50@router.get( 

51 "/{id}", 

52 summary="Get Wallet", 

53 response_model=CustomerWallet, 

54 responses={404: WalletNotFound}, 

55) 

56async def get( 

57 id: WalletID, 

58 auth_subject: auth.CustomerPortalRead, 

59 session: AsyncSession = Depends(get_db_session), 

60) -> Wallet: 

61 """Get a wallet by ID for the authenticated customer.""" 

62 wallet = await customer_wallet_service.get_by_id(session, auth_subject, id) 

63 

64 if wallet is None: 64 ↛ 67line 64 didn't jump to line 67 because the condition on line 64 was always true

65 raise ResourceNotFound() 

66 

67 return wallet