Coverage for polar/refund/endpoints.py: 88%
25 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.kit.pagination import ListResource, PaginationParamsQuery 1a
5from polar.kit.schemas import MultipleQueryFilter 1a
6from polar.models import Refund 1a
7from polar.openapi import APITag 1a
8from polar.order.schemas import OrderID 1a
9from polar.organization.schemas import OrganizationID 1a
10from polar.postgres import AsyncSession, get_db_session 1a
11from polar.routing import APIRouter 1a
12from polar.subscription.schemas import SubscriptionID 1a
14from . import auth 1a
15from .schemas import Refund as RefundSchema 1a
16from .schemas import RefundCreate, RefundID 1a
17from .service import RefundAmountTooHigh, RefundedAlready 1a
18from .service import refund as refund_service 1a
19from .sorting import RefundListSorting 1a
21router = APIRouter(prefix="/refunds", tags=["refunds", APITag.public]) 1a
24@router.get("/", summary="List Refunds", response_model=ListResource[RefundSchema]) 1a
25async def list( 1a
26 pagination: PaginationParamsQuery,
27 sorting: RefundListSorting,
28 auth_subject: auth.RefundsRead,
29 id: MultipleQueryFilter[RefundID] | None = Query(
30 None, title="RefundID Filter", description="Filter by refund ID."
31 ),
32 organization_id: MultipleQueryFilter[OrganizationID] | None = Query(
33 None, title="OrganizationID Filter", description="Filter by organization ID."
34 ),
35 order_id: MultipleQueryFilter[OrderID] | None = Query(
36 None, title="OrderID Filter", description="Filter by order ID."
37 ),
38 subscription_id: MultipleQueryFilter[SubscriptionID] | None = Query(
39 None, title="SubscriptionID Filter", description="Filter by subscription ID."
40 ),
41 customer_id: MultipleQueryFilter[CustomerID] | None = Query(
42 None, title="CustomerID Filter", description="Filter by customer ID."
43 ),
44 succeeded: bool | None = Query(
45 None,
46 title="RefundStatus Filter",
47 description="Filter by `succeeded`.",
48 ),
49 session: AsyncSession = Depends(get_db_session),
50) -> ListResource[RefundSchema]:
51 """List products."""
52 results, count = await refund_service.get_list(
53 session,
54 auth_subject,
55 id=id,
56 organization_id=organization_id,
57 order_id=order_id,
58 subscription_id=subscription_id,
59 customer_id=customer_id,
60 succeeded=succeeded,
61 pagination=pagination,
62 sorting=sorting,
63 )
65 return ListResource.from_paginated_results(
66 [RefundSchema.model_validate(result) for result in results],
67 count,
68 pagination,
69 )
72@router.post( 1a
73 "/",
74 summary="Create Refund",
75 response_model=RefundSchema,
76 responses={
77 201: {"description": "Refund created."},
78 400: {
79 "description": "Refund amount exceeds remaining order balance.",
80 "model": RefundAmountTooHigh.schema(),
81 },
82 403: {
83 "description": "Order is already fully refunded.",
84 "model": RefundedAlready.schema(),
85 },
86 },
87)
88async def create( 1a
89 refund_create: RefundCreate,
90 auth_subject: auth.RefundsWrite,
91 session: AsyncSession = Depends(get_db_session),
92) -> Refund:
93 """Create a refund."""
94 return await refund_service.user_create(
95 session,
96 auth_subject,
97 create_schema=refund_create,
98 )