Coverage for polar/transaction/endpoints.py: 72%
30 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, Query 1a
4from pydantic import UUID4 1a
6from polar.account.service import account as account_service 1a
7from polar.exceptions import ResourceNotFound 1a
8from polar.kit.pagination import ListResource, PaginationParamsQuery 1a
9from polar.kit.sorting import Sorting, SortingGetter 1a
10from polar.models.transaction import TransactionType 1a
11from polar.openapi import APITag 1a
12from polar.postgres import AsyncReadSession, get_db_read_session 1a
13from polar.routing import APIRouter 1a
14from polar.transaction import ( 1a
15 auth as transactions_auth,
16)
18from .schemas import Transaction, TransactionDetails, TransactionsSummary 1a
19from .service.transaction import TransactionSortProperty 1a
20from .service.transaction import transaction as transaction_service 1a
22router = APIRouter(prefix="/transactions", tags=["transactions", APITag.private]) 1a
25SearchSorting = Annotated[ 1a
26 list[Sorting[TransactionSortProperty]],
27 Depends(SortingGetter(TransactionSortProperty, ["-created_at"])),
28]
31@router.get("/search", response_model=ListResource[Transaction]) 1a
32async def search_transactions( 1a
33 pagination: PaginationParamsQuery,
34 sorting: SearchSorting,
35 auth_subject: transactions_auth.TransactionsRead,
36 type: TransactionType | None = Query(None),
37 account_id: UUID4 | None = Query(None),
38 payment_customer_id: UUID4 | None = Query(None),
39 payment_organization_id: UUID4 | None = Query(None),
40 payment_user_id: UUID4 | None = Query(None),
41 exclude_platform_fees: bool = Query(False),
42 session: AsyncReadSession = Depends(get_db_read_session),
43) -> ListResource[Transaction]:
44 results, count = await transaction_service.search(
45 session,
46 auth_subject.subject,
47 type=type,
48 account_id=account_id,
49 payment_customer_id=payment_customer_id,
50 payment_organization_id=payment_organization_id,
51 payment_user_id=payment_user_id,
52 exclude_platform_fees=exclude_platform_fees,
53 pagination=pagination,
54 sorting=sorting,
55 )
57 return ListResource.from_paginated_results(
58 [Transaction.model_validate(result) for result in results],
59 count,
60 pagination,
61 )
64@router.get("/lookup", response_model=TransactionDetails) 1a
65async def lookup_transaction( 1a
66 transaction_id: UUID4,
67 auth_subject: transactions_auth.TransactionsRead,
68 session: AsyncReadSession = Depends(get_db_read_session),
69) -> TransactionDetails:
70 return TransactionDetails.model_validate(
71 await transaction_service.lookup(session, transaction_id, auth_subject.subject)
72 )
75@router.get("/summary", response_model=TransactionsSummary) 1a
76async def get_summary( 1a
77 auth_subject: transactions_auth.TransactionsRead,
78 account_id: UUID4,
79 session: AsyncReadSession = Depends(get_db_read_session),
80) -> TransactionsSummary:
81 account = await account_service.get(session, auth_subject, account_id)
82 if account is None:
83 raise ResourceNotFound()
85 return await transaction_service.get_summary(session, account)