Coverage for polar/customer_portal/endpoints/order.py: 46%
68 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
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.schemas import MultipleQueryFilter 1a
9from polar.kit.sorting import Sorting, SortingGetter 1a
10from polar.models import Order 1a
11from polar.models.product import ProductBillingType 1a
12from polar.openapi import APITag 1a
13from polar.order.schemas import OrderID 1a
14from polar.order.service import ( 1a
15 MissingInvoiceBillingDetails,
16 NotPaidOrder,
17 PaymentAlreadyInProgress,
18)
19from polar.payment.repository import PaymentRepository 1a
20from polar.postgres import get_db_session 1a
21from polar.product.schemas import ProductID 1a
22from polar.routing import APIRouter 1a
23from polar.subscription.schemas import SubscriptionID 1a
25from .. import auth 1a
26from ..schemas.order import ( 1a
27 CustomerOrder,
28 CustomerOrderConfirmPayment,
29 CustomerOrderInvoice,
30 CustomerOrderPaymentConfirmation,
31 CustomerOrderPaymentStatus,
32 CustomerOrderUpdate,
33)
34from ..service.order import CustomerOrderSortProperty, OrderNotEligibleForRetry 1a
35from ..service.order import customer_order as customer_order_service 1a
37router = APIRouter(prefix="/orders", tags=["orders", APITag.public]) 1a
39OrderNotFound = {"description": "Order not found.", "model": ResourceNotFound.schema()} 1a
41ListSorting = Annotated[ 1a
42 list[Sorting[CustomerOrderSortProperty]],
43 Depends(SortingGetter(CustomerOrderSortProperty, ["-created_at"])),
44]
47@router.get("/", summary="List Orders", response_model=ListResource[CustomerOrder]) 1a
48async def list( 1a
49 auth_subject: auth.CustomerPortalRead,
50 pagination: PaginationParamsQuery,
51 sorting: ListSorting,
52 product_id: MultipleQueryFilter[ProductID] | None = Query(
53 None, title="ProductID Filter", description="Filter by product ID."
54 ),
55 product_billing_type: MultipleQueryFilter[ProductBillingType] | None = Query(
56 None,
57 title="ProductBillingType Filter",
58 description=(
59 "Filter by product billing type. "
60 "`recurring` will filter data corresponding "
61 "to subscriptions creations or renewals. "
62 "`one_time` will filter data corresponding to one-time purchases."
63 ),
64 ),
65 subscription_id: MultipleQueryFilter[SubscriptionID] | None = Query(
66 None, title="SubscriptionID Filter", description="Filter by subscription ID."
67 ),
68 query: str | None = Query(
69 None, description="Search by product or organization name."
70 ),
71 session: AsyncSession = Depends(get_db_session),
72) -> ListResource[CustomerOrder]:
73 """List orders of the authenticated customer."""
74 results, count = await customer_order_service.list(
75 session,
76 auth_subject,
77 product_id=product_id,
78 product_billing_type=product_billing_type,
79 subscription_id=subscription_id,
80 query=query,
81 pagination=pagination,
82 sorting=sorting,
83 )
85 return ListResource.from_paginated_results(
86 [CustomerOrder.model_validate(result) for result in results],
87 count,
88 pagination,
89 )
92@router.get( 1a
93 "/{id}",
94 summary="Get Order",
95 response_model=CustomerOrder,
96 responses={404: OrderNotFound},
97)
98async def get( 1a
99 id: OrderID,
100 auth_subject: auth.CustomerPortalRead,
101 session: AsyncSession = Depends(get_db_session),
102) -> Order:
103 """Get an order by ID for the authenticated customer."""
104 order = await customer_order_service.get_by_id(session, auth_subject, id)
106 if order is None:
107 raise ResourceNotFound()
109 return order
112@router.patch( 1a
113 "/{id}",
114 summary="Update Order",
115 response_model=CustomerOrder,
116 responses={404: OrderNotFound},
117)
118async def update( 1a
119 id: OrderID,
120 order_update: CustomerOrderUpdate,
121 auth_subject: auth.CustomerPortalWrite,
122 session: AsyncSession = Depends(get_db_session),
123) -> Order:
124 """Update an order for the authenticated customer."""
125 order = await customer_order_service.get_by_id(session, auth_subject, id)
127 if order is None:
128 raise ResourceNotFound()
130 return await customer_order_service.update(session, order, order_update)
133@router.post( 1a
134 "/{id}/invoice",
135 status_code=202,
136 summary="Generate Order Invoice",
137 responses={
138 422: {
139 "description": "Order is not paid or is missing billing name or address.",
140 "model": MissingInvoiceBillingDetails.schema() | NotPaidOrder.schema(),
141 },
142 },
143)
144async def generate_invoice( 1a
145 id: OrderID,
146 auth_subject: auth.CustomerPortalRead,
147 session: AsyncSession = Depends(get_db_session),
148) -> None:
149 """Trigger generation of an order's invoice."""
150 order = await customer_order_service.get_by_id(session, auth_subject, id)
152 if order is None:
153 raise ResourceNotFound()
155 await customer_order_service.trigger_invoice_generation(session, order)
158@router.get( 1a
159 "/{id}/invoice",
160 summary="Get Order Invoice",
161 response_model=CustomerOrderInvoice,
162 responses={404: OrderNotFound},
163)
164async def invoice( 1a
165 id: OrderID,
166 auth_subject: auth.CustomerPortalRead,
167 session: AsyncSession = Depends(get_db_session),
168) -> CustomerOrderInvoice:
169 """Get an order's invoice data."""
170 order = await customer_order_service.get_by_id(session, auth_subject, id)
172 if order is None:
173 raise ResourceNotFound()
175 return await customer_order_service.get_order_invoice(order)
178@router.get( 1a
179 "/{id}/payment-status",
180 summary="Get Order Payment Status",
181 response_model=CustomerOrderPaymentStatus,
182 responses={404: OrderNotFound},
183)
184async def get_payment_status( 1a
185 id: OrderID,
186 auth_subject: auth.CustomerPortalRead,
187 session: AsyncSession = Depends(get_db_session),
188) -> CustomerOrderPaymentStatus:
189 """Get the current payment status for an order."""
190 order = await customer_order_service.get_by_id(session, auth_subject, id)
192 if order is None:
193 raise ResourceNotFound()
195 payment_repository = PaymentRepository.from_session(session)
196 payment = await payment_repository.get_latest_for_order(order.id)
198 if payment is None:
199 return CustomerOrderPaymentStatus(
200 status="no_payment",
201 error=None,
202 )
204 return CustomerOrderPaymentStatus(
205 status=payment.status,
206 error=payment.decline_message if payment.decline_message is not None else None,
207 )
210@router.post( 1a
211 "/{id}/confirm-payment",
212 summary="Confirm Retry Payment",
213 response_model=CustomerOrderPaymentConfirmation,
214 responses={
215 404: OrderNotFound,
216 409: {
217 "description": "Payment already in progress.",
218 "model": PaymentAlreadyInProgress.schema(),
219 },
220 422: {
221 "description": "Order not eligible for retry or payment confirmation failed.",
222 "model": OrderNotEligibleForRetry.schema(),
223 },
224 },
225)
226async def confirm_retry_payment( 1a
227 id: OrderID,
228 confirm_data: CustomerOrderConfirmPayment,
229 auth_subject: auth.CustomerPortalWrite,
230 session: AsyncSession = Depends(get_db_session),
231) -> CustomerOrderPaymentConfirmation:
232 """Confirm a retry payment using a Stripe confirmation token."""
233 order = await customer_order_service.get_by_id(session, auth_subject, id)
235 if order is None:
236 raise ResourceNotFound()
238 return await customer_order_service.confirm_retry_payment(
239 session,
240 order,
241 confirm_data.confirmation_token_id,
242 confirm_data.payment_processor,
243 confirm_data.payment_method_id,
244 )