Coverage for polar/customer_portal/service/order.py: 40%
82 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 16:17 +0000
1import uuid 1a
2from collections.abc import Sequence 1a
3from enum import StrEnum 1a
4from typing import Any 1a
6from sqlalchemy import UnaryExpression, asc, desc 1a
7from sqlalchemy.orm import contains_eager 1a
9from polar.auth.models import AuthSubject 1a
10from polar.enums import PaymentProcessor 1a
11from polar.exceptions import PolarError 1a
12from polar.invoice.service import invoice as invoice_service 1a
13from polar.kit.db.postgres import AsyncSession 1a
14from polar.kit.pagination import PaginationParams 1a
15from polar.kit.sorting import Sorting 1a
16from polar.models import Customer, Order, Product 1a
17from polar.models.product import ProductBillingType 1a
18from polar.order.service import InvoiceDoesNotExist 1a
19from polar.order.service import order as order_service 1a
21from ..repository.order import CustomerOrderRepository 1a
22from ..schemas.order import ( 1a
23 CustomerOrderInvoice,
24 CustomerOrderPaymentConfirmation,
25 CustomerOrderUpdate,
26)
29class CustomerOrderError(PolarError): ... 1a
32class InvoiceNotAvailable(CustomerOrderError): 1a
33 def __init__(self, order: Order) -> None: 1a
34 self.order = order
35 message = "The invoice is not available for this order."
36 super().__init__(message, 404)
39class OrderNotEligibleForRetry(CustomerOrderError): 1a
40 def __init__(self, order: Order) -> None: 1a
41 self.order = order
42 message = "Order is not eligible for payment retry."
43 super().__init__(message, 422)
46class PaymentAlreadyInProgress(CustomerOrderError): 1a
47 def __init__(self, order: Order) -> None: 1a
48 self.order = order
49 message = "Payment for order is already in progress."
50 super().__init__(message, 409)
53class CustomerOrderSortProperty(StrEnum): 1a
54 created_at = "created_at" 1a
55 amount = "amount" 1a
56 net_amount = "net_amount" 1a
57 product = "product" 1a
58 subscription = "subscription" 1a
61class CustomerOrderService: 1a
62 async def list( 1a
63 self,
64 session: AsyncSession,
65 auth_subject: AuthSubject[Customer],
66 *,
67 product_id: Sequence[uuid.UUID] | None = None,
68 product_billing_type: Sequence[ProductBillingType] | None = None,
69 subscription_id: Sequence[uuid.UUID] | None = None,
70 query: str | None = None,
71 pagination: PaginationParams,
72 sorting: list[Sorting[CustomerOrderSortProperty]] = [
73 (CustomerOrderSortProperty.created_at, True)
74 ],
75 ) -> tuple[Sequence[Order], int]:
76 repository = CustomerOrderRepository.from_session(session)
77 statement = (
78 repository.get_readable_statement(auth_subject)
79 .join(Order.product, isouter=True)
80 .options(
81 *repository.get_eager_options(
82 product_load=contains_eager(Order.product)
83 )
84 )
85 )
87 if product_id is not None:
88 statement = statement.where(Order.product_id.in_(product_id))
90 if product_billing_type is not None:
91 statement = statement.where(Product.billing_type.in_(product_billing_type))
93 if subscription_id is not None:
94 statement = statement.where(Order.subscription_id.in_(subscription_id))
96 if query is not None:
97 statement = statement.where(Product.name.ilike(f"%{query}%"))
99 order_by_clauses: list[UnaryExpression[Any]] = []
100 for criterion, is_desc in sorting:
101 clause_function = desc if is_desc else asc
102 if criterion == CustomerOrderSortProperty.created_at:
103 order_by_clauses.append(clause_function(Order.created_at))
104 elif criterion in {
105 CustomerOrderSortProperty.amount,
106 CustomerOrderSortProperty.net_amount,
107 }:
108 order_by_clauses.append(clause_function(Order.net_amount))
109 elif criterion == CustomerOrderSortProperty.product:
110 order_by_clauses.append(clause_function(Product.name))
111 elif criterion == CustomerOrderSortProperty.subscription:
112 order_by_clauses.append(clause_function(Order.subscription_id))
113 statement = statement.order_by(*order_by_clauses)
115 return await repository.paginate(
116 statement, limit=pagination.limit, page=pagination.page
117 )
119 async def get_by_id( 1a
120 self,
121 session: AsyncSession,
122 auth_subject: AuthSubject[Customer],
123 id: uuid.UUID,
124 ) -> Order | None:
125 repository = CustomerOrderRepository.from_session(session)
126 statement = (
127 repository.get_readable_statement(auth_subject)
128 .where(Order.id == id)
129 .options(*repository.get_eager_options())
130 )
131 return await repository.get_one_or_none(statement)
133 async def update( 1a
134 self, session: AsyncSession, order: Order, order_update: CustomerOrderUpdate
135 ) -> Order:
136 return await order_service.update(session, order, order_update)
138 async def trigger_invoice_generation( 1a
139 self, session: AsyncSession, order: Order
140 ) -> None:
141 return await order_service.trigger_invoice_generation(session, order)
143 async def get_order_invoice(self, order: Order) -> CustomerOrderInvoice: 1a
144 if order.invoice_path is None:
145 raise InvoiceDoesNotExist(order)
147 url, _ = await invoice_service.get_order_invoice_url(order)
148 return CustomerOrderInvoice(url=url)
150 async def confirm_retry_payment( 1a
151 self,
152 session: AsyncSession,
153 order: Order,
154 confirmation_token_id: str | None,
155 payment_processor: PaymentProcessor,
156 payment_method_id: uuid.UUID | None = None,
157 ) -> CustomerOrderPaymentConfirmation:
158 return await order_service.process_retry_payment(
159 session, order, confirmation_token_id, payment_processor, payment_method_id
160 )
163customer_order = CustomerOrderService() 1a