Coverage for polar/order/repository.py: 27%
67 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 15:52 +0000
1from collections.abc import Sequence 1a
2from typing import TYPE_CHECKING, cast 1a
3from uuid import UUID 1a
5from sqlalchemy import CursorResult, Select, case, select, update 1a
6from sqlalchemy.orm import joinedload 1a
7from sqlalchemy.orm.strategy_options import selectinload 1a
9from polar.auth.models import ( 1a
10 AuthSubject,
11 Organization,
12 User,
13 is_customer,
14 is_organization,
15 is_user,
16)
17from polar.kit.repository import ( 1a
18 Options,
19 RepositoryBase,
20 RepositorySoftDeletionIDMixin,
21 RepositorySoftDeletionMixin,
22 RepositorySortingMixin,
23 SortingClause,
24)
25from polar.kit.utils import utc_now 1a
26from polar.models import ( 1a
27 Customer,
28 Discount,
29 Order,
30 OrderItem,
31 Product,
32 ProductPrice,
33 Subscription,
34 UserOrganization,
35)
36from polar.models.order import OrderStatus 1a
38from .sorting import OrderSortProperty 1a
40if TYPE_CHECKING: 40 ↛ 41line 40 didn't jump to line 41 because the condition on line 40 was never true1a
41 from sqlalchemy.orm.strategy_options import _AbstractLoad
44class OrderRepository( 1a
45 RepositorySortingMixin[Order, OrderSortProperty],
46 RepositorySoftDeletionIDMixin[Order, UUID],
47 RepositorySoftDeletionMixin[Order],
48 RepositoryBase[Order],
49):
50 model = Order 1a
52 async def get_all_by_customer( 1a
53 self,
54 customer_id: UUID,
55 *,
56 status: OrderStatus | None = None,
57 options: Options = (),
58 ) -> Sequence[Order]:
59 statement = (
60 self.get_base_statement()
61 .where(Order.customer_id == customer_id)
62 .options(*options)
63 )
64 if status is not None:
65 statement = statement.where(Order.status == status)
66 return await self.get_all(statement)
68 async def get_by_stripe_invoice_id( 1a
69 self, stripe_invoice_id: str, *, options: Options = ()
70 ) -> Order | None:
71 statement = (
72 self.get_base_statement()
73 .where(Order.stripe_invoice_id == stripe_invoice_id)
74 .options(*options)
75 )
76 return await self.get_one_or_none(statement)
78 async def get_earliest_by_checkout_id( 1a
79 self, checkout_id: UUID, *, options: Options = ()
80 ) -> Order | None:
81 statement = (
82 self.get_base_statement()
83 .where(Order.checkout_id == checkout_id)
84 .order_by(Order.created_at.asc())
85 .limit(1)
86 .options(*options)
87 )
88 return await self.get_one_or_none(statement)
90 async def get_due_dunning_orders(self, *, options: Options = ()) -> Sequence[Order]: 1a
91 """Get orders that are due for dunning retry based on next_payment_attempt_at."""
93 statement = (
94 self.get_base_statement()
95 .where(
96 Order.next_payment_attempt_at.is_not(None),
97 Order.next_payment_attempt_at <= utc_now(),
98 )
99 .order_by(Order.next_payment_attempt_at.asc())
100 .options(*options)
101 )
102 return await self.get_all(statement)
104 async def acquire_payment_lock_by_id(self, order_id: UUID) -> bool: 1a
105 """
106 Internal method to acquire a payment lock by order ID.
107 This is the original acquire_payment_lock logic.
109 Returns:
110 True if lock was acquired, False if already locked
111 """
112 statement = (
113 update(Order)
114 .where(Order.id == order_id, Order.payment_lock_acquired_at.is_(None))
115 .values(payment_lock_acquired_at=utc_now())
116 )
118 # https://github.com/sqlalchemy/sqlalchemy/commit/67f62aac5b49b6d048ca39019e5bd123d3c9cfb2
119 result = cast(CursorResult[Order], await self.session.execute(statement))
120 return result.rowcount > 0
122 async def release_payment_lock(self, order: Order, *, flush: bool = False) -> Order: 1a
123 """Release a payment lock for an order."""
124 return await self.update(
125 order, update_dict={"payment_lock_acquired_at": None}, flush=flush
126 )
128 def get_readable_statement( 1a
129 self, auth_subject: AuthSubject[User | Organization | Customer]
130 ) -> Select[tuple[Order]]:
131 statement = self.get_base_statement().join(
132 Customer, Order.customer_id == Customer.id
133 )
135 if is_user(auth_subject):
136 user = auth_subject.subject
137 statement = statement.where(
138 Customer.organization_id.in_(
139 select(UserOrganization.organization_id).where(
140 UserOrganization.user_id == user.id,
141 UserOrganization.deleted_at.is_(None),
142 )
143 )
144 )
145 elif is_organization(auth_subject):
146 statement = statement.where(
147 Customer.organization_id == auth_subject.subject.id,
148 )
149 elif is_customer(auth_subject):
150 customer = auth_subject.subject
151 statement = statement.where(
152 Order.customer_id == customer.id,
153 Order.deleted_at.is_(None),
154 )
156 return statement
158 def get_eager_options( 1a
159 self,
160 *,
161 customer_load: "_AbstractLoad | None" = None,
162 product_load: "_AbstractLoad | None" = None,
163 discount_load: "_AbstractLoad | None" = None,
164 ) -> Options:
165 return (
166 customer_load
167 if customer_load
168 else joinedload(Order.customer).joinedload(Customer.organization),
169 discount_load if discount_load else joinedload(Order.discount),
170 product_load if product_load else joinedload(Order.product),
171 joinedload(Order.subscription).joinedload(Subscription.customer),
172 selectinload(Order.items)
173 .joinedload(OrderItem.product_price)
174 .joinedload(ProductPrice.product),
175 )
177 def get_sorting_clause(self, property: OrderSortProperty) -> SortingClause: 1a
178 match property:
179 case OrderSortProperty.created_at:
180 return Order.created_at
181 case OrderSortProperty.status:
182 return case(
183 (Order.status == OrderStatus.pending, 1),
184 (Order.status == OrderStatus.paid, 2),
185 (Order.status == OrderStatus.refunded, 3),
186 (Order.status == OrderStatus.partially_refunded, 4),
187 )
188 case OrderSortProperty.invoice_number:
189 return Order.invoice_number
190 case OrderSortProperty.amount | OrderSortProperty.net_amount:
191 return Order.net_amount
192 case OrderSortProperty.customer:
193 return Customer.email
194 case OrderSortProperty.product:
195 return Product.name
196 case OrderSortProperty.discount:
197 return Discount.name
198 case OrderSortProperty.subscription:
199 return Order.subscription_id