Coverage for polar/order/endpoints.py: 86%
61 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 12:42 +0000
1from collections.abc import AsyncGenerator
3from fastapi import Depends, Query, Response
4from fastapi.responses import StreamingResponse
5from pydantic import UUID4
7from polar.customer.schemas.customer import CustomerID
8from polar.exceptions import ResourceNotFound
9from polar.kit.csv import IterableCSVWriter
10from polar.kit.metadata import MetadataQuery, get_metadata_query_openapi_schema
11from polar.kit.pagination import ListResource, PaginationParams, PaginationParamsQuery
12from polar.kit.schemas import MultipleQueryFilter
13from polar.models import Order
14from polar.models.product import ProductBillingType
15from polar.openapi import APITag
16from polar.organization.schemas import OrganizationID
17from polar.postgres import (
18 AsyncReadSession,
19 AsyncSession,
20 get_db_read_session,
21 get_db_session,
22)
23from polar.product.schemas import ProductID
24from polar.routing import APIRouter
26from . import auth, sorting
27from .schemas import Order as OrderSchema
28from .schemas import OrderID, OrderInvoice, OrderNotFound, OrderUpdate
29from .service import MissingInvoiceBillingDetails, NotPaidOrder
30from .service import order as order_service
32router = APIRouter(prefix="/orders", tags=["orders", APITag.public, APITag.mcp])
35@router.get(
36 "/",
37 summary="List Orders",
38 response_model=ListResource[OrderSchema],
39 openapi_extra={"parameters": [get_metadata_query_openapi_schema()]},
40)
41async def list(
42 auth_subject: auth.OrdersRead,
43 pagination: PaginationParamsQuery,
44 sorting: sorting.ListSorting,
45 metadata: MetadataQuery,
46 organization_id: MultipleQueryFilter[OrganizationID] | None = Query(
47 None, title="OrganizationID Filter", description="Filter by organization ID."
48 ),
49 product_id: MultipleQueryFilter[ProductID] | None = Query(
50 None, title="ProductID Filter", description="Filter by product ID."
51 ),
52 product_billing_type: MultipleQueryFilter[ProductBillingType] | None = Query(
53 None,
54 title="ProductBillingType Filter",
55 description=(
56 "Filter by product billing type. "
57 "`recurring` will filter data corresponding "
58 "to subscriptions creations or renewals. "
59 "`one_time` will filter data corresponding to one-time purchases."
60 ),
61 ),
62 discount_id: MultipleQueryFilter[UUID4] | None = Query(
63 None, title="DiscountID Filter", description="Filter by discount ID."
64 ),
65 customer_id: MultipleQueryFilter[CustomerID] | None = Query(
66 None, title="CustomerID Filter", description="Filter by customer ID."
67 ),
68 checkout_id: MultipleQueryFilter[UUID4] | None = Query(
69 None, title="CheckoutID Filter", description="Filter by checkout ID."
70 ),
71 session: AsyncReadSession = Depends(get_db_read_session),
72) -> ListResource[OrderSchema]:
73 """List orders."""
74 results, count = await order_service.list(
75 session,
76 auth_subject,
77 organization_id=organization_id,
78 product_id=product_id,
79 product_billing_type=product_billing_type,
80 discount_id=discount_id,
81 customer_id=customer_id,
82 checkout_id=checkout_id,
83 metadata=metadata,
84 pagination=pagination,
85 sorting=sorting,
86 )
88 return ListResource.from_paginated_results(
89 [OrderSchema.model_validate(result) for result in results],
90 count,
91 pagination,
92 )
95@router.get("/export", summary="Export Subscriptions")
96async def export(
97 auth_subject: auth.OrdersRead,
98 organization_id: MultipleQueryFilter[OrganizationID] | None = Query(
99 None, title="OrganizationID Filter", description="Filter by organization ID."
100 ),
101 product_id: MultipleQueryFilter[ProductID] | None = Query(
102 None, title="ProductID Filter", description="Filter by product ID."
103 ),
104 session: AsyncReadSession = Depends(get_db_read_session),
105) -> Response:
106 """Export orders as a CSV file."""
108 async def create_csv() -> AsyncGenerator[str, None]:
109 csv_writer = IterableCSVWriter(dialect="excel")
110 # CSV header
111 yield csv_writer.getrow(
112 (
113 "Email",
114 "Created At",
115 "Product",
116 "Amount",
117 "Currency",
118 "Status",
119 "Invoice number",
120 )
121 )
123 (results, _) = await order_service.list(
124 session,
125 auth_subject,
126 organization_id=organization_id,
127 product_id=product_id,
128 pagination=PaginationParams(limit=1000000, page=1),
129 )
131 for order in results: 131 ↛ 132line 131 didn't jump to line 132 because the loop on line 131 never started
132 yield csv_writer.getrow(
133 (
134 order.customer.email,
135 order.created_at.isoformat(),
136 order.description,
137 order.net_amount / 100,
138 order.currency,
139 order.status,
140 order.invoice_number,
141 )
142 )
144 filename = "polar-orders.csv"
145 return StreamingResponse(
146 create_csv(),
147 media_type="text/csv",
148 headers={"Content-Disposition": f"attachment; filename={filename}"},
149 )
152@router.get(
153 "/{id}",
154 summary="Get Order",
155 response_model=OrderSchema,
156 responses={404: OrderNotFound},
157)
158async def get(
159 id: OrderID,
160 auth_subject: auth.OrdersRead,
161 session: AsyncReadSession = Depends(get_db_read_session),
162) -> Order:
163 """Get an order by ID."""
164 order = await order_service.get(session, auth_subject, id)
166 if order is None: 166 ↛ 169line 166 didn't jump to line 169 because the condition on line 166 was always true
167 raise ResourceNotFound()
169 return order
172@router.patch(
173 "/{id}",
174 summary="Update Order",
175 response_model=OrderSchema,
176 responses={404: OrderNotFound},
177)
178async def update(
179 id: OrderID,
180 order_update: OrderUpdate,
181 auth_subject: auth.OrdersWrite,
182 session: AsyncSession = Depends(get_db_session),
183) -> Order:
184 """Update an order."""
185 order = await order_service.get(session, auth_subject, id)
187 if order is None: 187 ↛ 190line 187 didn't jump to line 190 because the condition on line 187 was always true
188 raise ResourceNotFound()
190 return await order_service.update(session, order, order_update)
193@router.post(
194 "/{id}/invoice",
195 status_code=202,
196 summary="Generate Order Invoice",
197 responses={
198 422: {
199 "description": "Order is not paid or is missing billing name or address.",
200 "model": MissingInvoiceBillingDetails.schema() | NotPaidOrder.schema(),
201 },
202 },
203)
204async def generate_invoice(
205 id: OrderID,
206 auth_subject: auth.OrdersRead,
207 session: AsyncSession = Depends(get_db_session),
208) -> None:
209 """Trigger generation of an order's invoice."""
210 order = await order_service.get(session, auth_subject, id)
212 if order is None: 212 ↛ 215line 212 didn't jump to line 215 because the condition on line 212 was always true
213 raise ResourceNotFound()
215 await order_service.trigger_invoice_generation(session, order)
218@router.get(
219 "/{id}/invoice",
220 summary="Get Order Invoice",
221 response_model=OrderInvoice,
222 responses={404: OrderNotFound},
223)
224async def invoice(
225 id: OrderID,
226 auth_subject: auth.OrdersRead,
227 session: AsyncReadSession = Depends(get_db_read_session),
228) -> OrderInvoice:
229 """Get an order's invoice data."""
230 order = await order_service.get(session, auth_subject, id)
232 if order is None: 232 ↛ 235line 232 didn't jump to line 235 because the condition on line 232 was always true
233 raise ResourceNotFound()
235 return await order_service.get_order_invoice(order)