Coverage for polar/customer_portal/endpoints/order.py: 78%
68 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 12:42 +0000
1from typing import Annotated
3from fastapi import Depends, Query
5from polar.exceptions import ResourceNotFound
6from polar.kit.db.postgres import AsyncSession
7from polar.kit.pagination import ListResource, PaginationParamsQuery
8from polar.kit.schemas import MultipleQueryFilter
9from polar.kit.sorting import Sorting, SortingGetter
10from polar.models import Order
11from polar.models.product import ProductBillingType
12from polar.openapi import APITag
13from polar.order.schemas import OrderID
14from polar.order.service import (
15 MissingInvoiceBillingDetails,
16 NotPaidOrder,
17 PaymentAlreadyInProgress,
18)
19from polar.payment.repository import PaymentRepository
20from polar.postgres import get_db_session
21from polar.product.schemas import ProductID
22from polar.routing import APIRouter
23from polar.subscription.schemas import SubscriptionID
25from .. import auth
26from ..schemas.order import (
27 CustomerOrder,
28 CustomerOrderConfirmPayment,
29 CustomerOrderInvoice,
30 CustomerOrderPaymentConfirmation,
31 CustomerOrderPaymentStatus,
32 CustomerOrderUpdate,
33)
34from ..service.order import CustomerOrderSortProperty, OrderNotEligibleForRetry
35from ..service.order import customer_order as customer_order_service
37router = APIRouter(prefix="/orders", tags=["orders", APITag.public])
39OrderNotFound = {"description": "Order not found.", "model": ResourceNotFound.schema()}
41ListSorting = Annotated[
42 list[Sorting[CustomerOrderSortProperty]],
43 Depends(SortingGetter(CustomerOrderSortProperty, ["-created_at"])),
44]
47@router.get("/", summary="List Orders", response_model=ListResource[CustomerOrder])
48async def list(
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(
93 "/{id}",
94 summary="Get Order",
95 response_model=CustomerOrder,
96 responses={404: OrderNotFound},
97)
98async def get(
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: 106 ↛ 109line 106 didn't jump to line 109 because the condition on line 106 was always true
107 raise ResourceNotFound()
109 return order
112@router.patch(
113 "/{id}",
114 summary="Update Order",
115 response_model=CustomerOrder,
116 responses={404: OrderNotFound},
117)
118async def update(
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: 127 ↛ 130line 127 didn't jump to line 130 because the condition on line 127 was always true
128 raise ResourceNotFound()
130 return await customer_order_service.update(session, order, order_update)
133@router.post(
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(
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: 152 ↛ 155line 152 didn't jump to line 155 because the condition on line 152 was always true
153 raise ResourceNotFound()
155 await customer_order_service.trigger_invoice_generation(session, order)
158@router.get(
159 "/{id}/invoice",
160 summary="Get Order Invoice",
161 response_model=CustomerOrderInvoice,
162 responses={404: OrderNotFound},
163)
164async def invoice(
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: 172 ↛ 175line 172 didn't jump to line 175 because the condition on line 172 was always true
173 raise ResourceNotFound()
175 return await customer_order_service.get_order_invoice(order)
178@router.get(
179 "/{id}/payment-status",
180 summary="Get Order Payment Status",
181 response_model=CustomerOrderPaymentStatus,
182 responses={404: OrderNotFound},
183)
184async def get_payment_status(
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: 192 ↛ 195line 192 didn't jump to line 195 because the condition on line 192 was always true
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(
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(
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: 235 ↛ 238line 235 didn't jump to line 238 because the condition on line 235 was always true
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 )