Coverage for polar/customer_portal/service/order.py: 81%

1import uuid 

2from collections.abc import Sequence 

3from enum import StrEnum 

4from typing import Any 

5 

6from sqlalchemy import UnaryExpression, asc, desc 

7from sqlalchemy.orm import contains_eager 

8 

9from polar.auth.models import AuthSubject 

10from polar.enums import PaymentProcessor 

11from polar.exceptions import PolarError 

12from polar.invoice.service import invoice as invoice_service 

13from polar.kit.db.postgres import AsyncSession 

14from polar.kit.pagination import PaginationParams 

15from polar.kit.sorting import Sorting 

16from polar.models import Customer, Order, Product 

17from polar.models.product import ProductBillingType 

18from polar.order.service import InvoiceDoesNotExist 

19from polar.order.service import order as order_service 

20 

21from ..repository.order import CustomerOrderRepository 

22from ..schemas.order import ( 

23 CustomerOrderInvoice, 

24 CustomerOrderPaymentConfirmation, 

25 CustomerOrderUpdate, 

26) 

27 

28 

29class CustomerOrderError(PolarError): ... 29 ↛ 32line 29 didn't jump to line 32 because

30 

31 

32class InvoiceNotAvailable(CustomerOrderError): 

33 def __init__(self, order: Order) -> None: 

34 self.order = order 

35 message = "The invoice is not available for this order." 

36 super().__init__(message, 404) 

37 

38 

39class OrderNotEligibleForRetry(CustomerOrderError): 

40 def __init__(self, order: Order) -> None: 

41 self.order = order 

42 message = "Order is not eligible for payment retry." 

43 super().__init__(message, 422) 

44 

45 

46class PaymentAlreadyInProgress(CustomerOrderError): 

47 def __init__(self, order: Order) -> None: 

48 self.order = order 

49 message = "Payment for order is already in progress." 

50 super().__init__(message, 409) 

51 

52 

53class CustomerOrderSortProperty(StrEnum): 

54 created_at = "created_at" 

55 amount = "amount" 

56 net_amount = "net_amount" 

57 product = "product" 

58 subscription = "subscription" 

59 

60 

61class CustomerOrderService: 

62 async def list( 

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 ) 

86 

87 if product_id is not None: 

88 statement = statement.where(Order.product_id.in_(product_id)) 

89 

90 if product_billing_type is not None: 

91 statement = statement.where(Product.billing_type.in_(product_billing_type)) 

92 

93 if subscription_id is not None: 

94 statement = statement.where(Order.subscription_id.in_(subscription_id)) 

95 

96 if query is not None: 

97 statement = statement.where(Product.name.ilike(f"%{query}%")) 

98 

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: 111 ↛ 100line 111 didn't jump to line 100 because the condition on line 111 was always true

112 order_by_clauses.append(clause_function(Order.subscription_id)) 

113 statement = statement.order_by(*order_by_clauses) 

114 

115 return await repository.paginate( 

116 statement, limit=pagination.limit, page=pagination.page 

117 ) 

118 

119 async def get_by_id( 

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) 

132 

133 async def update( 

134 self, session: AsyncSession, order: Order, order_update: CustomerOrderUpdate 

135 ) -> Order: 

136 return await order_service.update(session, order, order_update) 

137 

138 async def trigger_invoice_generation( 

139 self, session: AsyncSession, order: Order 

140 ) -> None: 

141 return await order_service.trigger_invoice_generation(session, order) 

142 

143 async def get_order_invoice(self, order: Order) -> CustomerOrderInvoice: 

144 if order.invoice_path is None: 

145 raise InvoiceDoesNotExist(order) 

146 

147 url, _ = await invoice_service.get_order_invoice_url(order) 

148 return CustomerOrderInvoice(url=url) 

149 

150 async def confirm_retry_payment( 

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 ) 

161 

162 

163customer_order = CustomerOrderService()