Coverage for polar/payment/endpoints.py: 72%
27 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 17:15 +0000
1from fastapi import Depends, Query 1a
2from pydantic.types import UUID4 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 Payment 1a
8from polar.models.payment import PaymentStatus 1a
9from polar.openapi import APITag 1a
10from polar.organization.schemas import OrganizationID 1a
11from polar.postgres import AsyncReadSession, get_db_read_session 1a
12from polar.routing import APIRouter 1a
14from . import auth, sorting 1a
15from .schemas import Payment as PaymentSchema 1a
16from .schemas import PaymentAdapter, PaymentID 1a
17from .service import payment as payment_service 1a
19router = APIRouter(prefix="/payments", tags=["payments", APITag.public, APITag.mcp]) 1a
22PaymentNotFound = { 1a
23 "description": "Payment not found.",
24 "model": ResourceNotFound.schema(),
25}
28@router.get("/", summary="List Payments", response_model=ListResource[PaymentSchema]) 1a
29async def list( 1a
30 auth_subject: auth.PaymentRead,
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 checkout_id: MultipleQueryFilter[UUID4] | None = Query(
37 None, title="CheckoutID Filter", description="Filter by checkout ID."
38 ),
39 order_id: MultipleQueryFilter[UUID4] | None = Query(
40 None, title="OrderID Filter", description="Filter by order ID."
41 ),
42 status: MultipleQueryFilter[PaymentStatus] | None = Query(
43 None, title="Status Filter", description="Filter by payment status."
44 ),
45 method: MultipleQueryFilter[str] | None = Query(
46 None, title="Method Filter", description="Filter by payment method."
47 ),
48 customer_email: MultipleQueryFilter[str] | None = Query(
49 None, title="CustomerEmail Filter", description="Filter by customer email."
50 ),
51 session: AsyncReadSession = Depends(get_db_read_session),
52) -> ListResource[PaymentSchema]:
53 """List payments."""
54 results, count = await payment_service.list(
55 session,
56 auth_subject,
57 organization_id=organization_id,
58 checkout_id=checkout_id,
59 order_id=order_id,
60 status=status,
61 method=method,
62 customer_email=customer_email,
63 pagination=pagination,
64 sorting=sorting,
65 )
67 return ListResource.from_paginated_results(
68 [PaymentAdapter.validate_python(result) for result in results],
69 count,
70 pagination,
71 )
74@router.get( 1a
75 "/{id}",
76 summary="Get Payment",
77 response_model=PaymentSchema,
78 responses={404: PaymentNotFound},
79)
80async def get( 1a
81 id: PaymentID,
82 auth_subject: auth.PaymentRead,
83 session: AsyncReadSession = Depends(get_db_read_session),
84) -> Payment:
85 """Get a payment by ID."""
86 payment = await payment_service.get(session, auth_subject, id)
88 if payment is None:
89 raise ResourceNotFound()
91 return payment