Coverage for polar/wallet/endpoints.py: 87%

1from fastapi import Depends, Query 

2 

3from polar.customer.schemas.customer import CustomerID 

4from polar.exceptions import ResourceNotFound 

5from polar.kit.pagination import ListResource, PaginationParamsQuery 

6from polar.kit.schemas import MultipleQueryFilter 

7from polar.models import Wallet 

8from polar.models.wallet import WalletType 

9from polar.openapi import APITag 

10from polar.organization.schemas import OrganizationID 

11from polar.postgres import ( 

12 AsyncReadSession, 

13 AsyncSession, 

14 get_db_read_session, 

15 get_db_session, 

16) 

17from polar.routing import APIRouter 

18 

19from . import auth, sorting 

20from .schemas import Wallet as WalletSchema 

21from .schemas import WalletID, WalletNotFound, WalletTopUpCreate 

22from .service import MissingPaymentMethodError, PaymentIntentFailedError 

23from .service import wallet as wallet_service 

24 

25router = APIRouter(prefix="/wallets", tags=["wallets", APITag.private]) 

26 

27 

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

29async def list( 

30 auth_subject: auth.WalletsRead, 

31 pagination: PaginationParamsQuery, 

32 sorting: sorting.ListSorting, 

33 organization_id: MultipleQueryFilter[OrganizationID] | None = Query( 

34 None, title="OrganizationID Filter", description="Filter by organization ID." 

35 ), 

36 type: MultipleQueryFilter[WalletType] | None = Query( 

37 None, title="Wallet Type Filter", description="Filter by wallet type." 

38 ), 

39 customer_id: MultipleQueryFilter[CustomerID] | None = Query( 

40 None, title="CustomerID Filter", description="Filter by customer ID." 

41 ), 

42 session: AsyncReadSession = Depends(get_db_read_session), 

43) -> ListResource[WalletSchema]: 

44 """List wallets.""" 

45 results, count = await wallet_service.list( 

46 session, 

47 auth_subject, 

48 organization_id=organization_id, 

49 type=type, 

50 customer_id=customer_id, 

51 pagination=pagination, 

52 sorting=sorting, 

53 ) 

54 

55 return ListResource.from_paginated_results( 

56 [WalletSchema.model_validate(result) for result in results], 

57 count, 

58 pagination, 

59 ) 

60 

61 

62@router.get( 

63 "/{id}", 

64 summary="Get Wallet", 

65 response_model=WalletSchema, 

66 responses={404: WalletNotFound}, 

67) 

68async def get( 

69 id: WalletID, 

70 auth_subject: auth.WalletsRead, 

71 session: AsyncReadSession = Depends(get_db_read_session), 

72) -> Wallet: 

73 """Get a wallet by ID.""" 

74 wallet = await wallet_service.get(session, auth_subject, id) 

75 

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

77 raise ResourceNotFound() 

78 

79 return wallet 

80 

81 

82@router.post( 

83 "/{id}/top-up", 

84 summary="Top-Up Wallet", 

85 response_model=WalletSchema, 

86 responses={ 

87 201: {"description": "Wallet topped up successfully."}, 

88 400: { 

89 "description": "The payment request failed.", 

90 "model": PaymentIntentFailedError.schema(), 

91 }, 

92 404: WalletNotFound, 

93 402: { 

94 "description": "No payment method available.", 

95 "model": MissingPaymentMethodError.schema(), 

96 }, 

97 }, 

98) 

99async def top_up( 

100 id: WalletID, 

101 top_up_create: WalletTopUpCreate, 

102 auth_subject: auth.WalletsWrite, 

103 session: AsyncSession = Depends(get_db_session), 

104) -> Wallet: 

105 """ 

106 Top-up a wallet by adding funds to its balance. 

107 

108 The customer should have a valid payment method on file. 

109 """ 

110 wallet = await wallet_service.get(session, auth_subject, id) 

111 

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

113 raise ResourceNotFound() 

114 

115 await wallet_service.top_up(session, wallet, top_up_create.amount) 

116 

117 return wallet