Coverage for polar/wallet/endpoints.py: 61%
34 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 17:15 +0000
1from fastapi import Depends, Query 1a
3from polar.customer.schemas.customer import CustomerID 1a
4from polar.exceptions import ResourceNotFound 1a
5from polar.kit.pagination import ListResource, PaginationParamsQuery 1a
6from polar.kit.schemas import MultipleQueryFilter 1a
7from polar.models import Wallet 1a
8from polar.models.wallet import WalletType 1a
9from polar.openapi import APITag 1a
10from polar.organization.schemas import OrganizationID 1a
11from polar.postgres import ( 1a
12 AsyncReadSession,
13 AsyncSession,
14 get_db_read_session,
15 get_db_session,
16)
17from polar.routing import APIRouter 1a
19from . import auth, sorting 1a
20from .schemas import Wallet as WalletSchema 1a
21from .schemas import WalletID, WalletNotFound, WalletTopUpCreate 1a
22from .service import MissingPaymentMethodError, PaymentIntentFailedError 1a
23from .service import wallet as wallet_service 1a
25router = APIRouter(prefix="/wallets", tags=["wallets", APITag.private]) 1a
28@router.get("/", summary="List Wallets", response_model=ListResource[WalletSchema]) 1a
29async def list( 1a
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 )
55 return ListResource.from_paginated_results(
56 [WalletSchema.model_validate(result) for result in results],
57 count,
58 pagination,
59 )
62@router.get( 1a
63 "/{id}",
64 summary="Get Wallet",
65 response_model=WalletSchema,
66 responses={404: WalletNotFound},
67)
68async def get( 1a
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)
76 if wallet is None:
77 raise ResourceNotFound()
79 return wallet
82@router.post( 1a
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( 1a
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.
108 The customer should have a valid payment method on file.
109 """
110 wallet = await wallet_service.get(session, auth_subject, id)
112 if wallet is None:
113 raise ResourceNotFound()
115 await wallet_service.top_up(session, wallet, top_up_create.amount)
117 return wallet