Coverage for polar/customer_portal/endpoints/wallet.py: 72%
27 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 17:15 +0000
1from typing import Annotated 1a
3from fastapi import Depends 1a
5from polar.exceptions import ResourceNotFound 1a
6from polar.kit.db.postgres import AsyncSession 1a
7from polar.kit.pagination import ListResource, PaginationParamsQuery 1a
8from polar.kit.sorting import Sorting, SortingGetter 1a
9from polar.models import Wallet 1a
10from polar.openapi import APITag 1a
11from polar.postgres import get_db_session 1a
12from polar.routing import APIRouter 1a
13from polar.wallet.schemas import WalletID, WalletNotFound 1a
15from .. import auth 1a
16from ..schemas.wallet import CustomerWallet 1a
17from ..service.wallet import customer_wallet as customer_wallet_service 1a
18from ..sorting.wallet import CustomerWalletSortProperty 1a
20router = APIRouter(prefix="/wallets", tags=["wallets", APITag.public]) 1a
22ListSorting = Annotated[ 1a
23 list[Sorting[CustomerWalletSortProperty]],
24 Depends(SortingGetter(CustomerWalletSortProperty, ["-created_at"])),
25]
28@router.get("/", summary="List Wallets", response_model=ListResource[CustomerWallet]) 1a
29async def list( 1a
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 )
43 return ListResource.from_paginated_results(
44 [CustomerWallet.model_validate(result) for result in results],
45 count,
46 pagination,
47 )
50@router.get( 1a
51 "/{id}",
52 summary="Get Wallet",
53 response_model=CustomerWallet,
54 responses={404: WalletNotFound},
55)
56async def get( 1a
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)
64 if wallet is None:
65 raise ResourceNotFound()
67 return wallet