Total Complexity | 60 |
Total Lines | 721 |
Duplicated Lines | 5.55 % |
Coverage | 69.06% |
Changes | 0 |
Duplicate code is one of the most pungent code smells. A rule that is often used is to re-structure code once it is duplicated in three or more places.
Common duplication problems, and corresponding solutions are:
Complex classes like byceps.services.shop.order.service often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.
Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.
1 | """ |
||
2 | byceps.services.shop.order.service |
||
3 | ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ |
||
4 | |||
5 | :Copyright: 2006-2021 Jochen Kupperschmidt |
||
6 | :License: Revised BSD (see `LICENSE` file for details) |
||
7 | """ |
||
8 | |||
9 | 1 | from __future__ import annotations |
|
10 | 1 | from datetime import datetime |
|
11 | 1 | from typing import Iterator, Mapping, Optional, Sequence |
|
12 | |||
13 | 1 | from flask import current_app |
|
14 | 1 | from flask_babel import lazy_gettext |
|
15 | 1 | from sqlalchemy.exc import IntegrityError |
|
16 | |||
17 | 1 | from ....database import db, paginate, Pagination |
|
18 | 1 | from ....events.shop import ShopOrderCanceled, ShopOrderPaid, ShopOrderPlaced |
|
19 | 1 | from ....typing import UserID |
|
20 | |||
21 | 1 | from ...user import service as user_service |
|
22 | |||
23 | 1 | from ..article import service as article_service |
|
24 | 1 | from ..cart.models import Cart |
|
25 | 1 | from ..shop.dbmodels import Shop as DbShop |
|
26 | 1 | from ..shop import service as shop_service |
|
27 | 1 | from ..shop.transfer.models import ShopID |
|
28 | 1 | from ..storefront import service as storefront_service |
|
29 | 1 | from ..storefront.transfer.models import StorefrontID |
|
30 | |||
31 | 1 | from .dbmodels.order import Order as DbOrder |
|
32 | 1 | from .dbmodels.order_event import OrderEvent as DbOrderEvent, OrderEventData |
|
33 | 1 | from .dbmodels.order_item import OrderItem as DbOrderItem |
|
34 | 1 | from .models.orderer import Orderer |
|
35 | 1 | from . import action_service, event_service, sequence_service |
|
36 | 1 | from .transfer.models import ( |
|
37 | Address, |
||
38 | Order, |
||
39 | OrderID, |
||
40 | OrderItem, |
||
41 | OrderNumber, |
||
42 | OrderState, |
||
43 | PaymentMethod, |
||
44 | PaymentState, |
||
45 | ) |
||
46 | |||
47 | |||
48 | 1 | class OrderFailed(Exception): |
|
49 | 1 | pass |
|
50 | |||
51 | |||
52 | 1 | def place_order( |
|
53 | storefront_id: StorefrontID, |
||
54 | orderer: Orderer, |
||
55 | cart: Cart, |
||
56 | *, |
||
57 | created_at: Optional[datetime] = None, |
||
58 | ) -> tuple[Order, ShopOrderPlaced]: |
||
59 | """Place an order for one or more articles.""" |
||
60 | 1 | storefront = storefront_service.get_storefront(storefront_id) |
|
61 | 1 | shop = shop_service.get_shop(storefront.shop_id) |
|
62 | |||
63 | 1 | orderer_user = user_service.get_user(orderer.user_id) |
|
64 | |||
65 | 1 | order_number_sequence = sequence_service.get_order_number_sequence( |
|
66 | storefront.order_number_sequence_id |
||
67 | ) |
||
68 | 1 | order_number = sequence_service.generate_order_number( |
|
69 | order_number_sequence.id |
||
70 | ) |
||
71 | |||
72 | 1 | order = _build_order(shop.id, order_number, orderer, created_at) |
|
73 | 1 | order_items = list(_build_order_items(cart, order)) |
|
74 | 1 | order.total_amount = cart.calculate_total_amount() |
|
75 | 1 | order.shipping_required = any( |
|
76 | item.shipping_required for item in order_items |
||
77 | ) |
||
78 | |||
79 | 1 | db.session.add(order) |
|
80 | 1 | db.session.add_all(order_items) |
|
81 | |||
82 | 1 | _reduce_article_stock(cart) |
|
83 | |||
84 | 1 | try: |
|
85 | 1 | db.session.commit() |
|
86 | except IntegrityError as e: |
||
87 | current_app.logger.error('Order %s failed: %s', order_number, e) |
||
88 | db.session.rollback() |
||
89 | raise OrderFailed() |
||
90 | |||
91 | 1 | order_dto = _order_to_transfer_object(order) |
|
92 | |||
93 | 1 | event = ShopOrderPlaced( |
|
94 | occurred_at=order.created_at, |
||
95 | initiator_id=orderer_user.id, |
||
96 | initiator_screen_name=orderer_user.screen_name, |
||
97 | order_id=order.id, |
||
98 | order_number=order.order_number, |
||
99 | orderer_id=orderer_user.id, |
||
100 | orderer_screen_name=orderer_user.screen_name, |
||
101 | ) |
||
102 | |||
103 | 1 | return order_dto, event |
|
104 | |||
105 | |||
106 | 1 | def _build_order( |
|
107 | shop_id: ShopID, |
||
108 | order_number: OrderNumber, |
||
109 | orderer: Orderer, |
||
110 | created_at: Optional[datetime], |
||
111 | ) -> DbOrder: |
||
112 | """Build an order.""" |
||
113 | 1 | return DbOrder( |
|
114 | shop_id, |
||
115 | order_number, |
||
116 | orderer.user_id, |
||
117 | orderer.first_names, |
||
118 | orderer.last_name, |
||
119 | orderer.country, |
||
120 | orderer.zip_code, |
||
121 | orderer.city, |
||
122 | orderer.street, |
||
123 | created_at=created_at, |
||
124 | ) |
||
125 | |||
126 | |||
127 | 1 | def _build_order_items(cart: Cart, order: DbOrder) -> Iterator[DbOrderItem]: |
|
128 | """Build order items from the cart's content.""" |
||
129 | 1 | for cart_item in cart.get_items(): |
|
130 | 1 | article = cart_item.article |
|
131 | 1 | quantity = cart_item.quantity |
|
132 | 1 | line_amount = cart_item.line_amount |
|
133 | |||
134 | 1 | yield DbOrderItem( |
|
135 | order, |
||
136 | article.item_number, |
||
137 | article.description, |
||
138 | article.price, |
||
139 | article.tax_rate, |
||
140 | quantity, |
||
141 | line_amount, |
||
142 | article.shipping_required, |
||
143 | ) |
||
144 | |||
145 | |||
146 | 1 | def _reduce_article_stock(cart: Cart) -> None: |
|
147 | """Reduce article stock according to what is in the cart.""" |
||
148 | 1 | for cart_item in cart.get_items(): |
|
149 | 1 | article = cart_item.article |
|
150 | 1 | quantity = cart_item.quantity |
|
151 | |||
152 | 1 | article_service.decrease_quantity(article.id, quantity, commit=False) |
|
153 | |||
154 | |||
155 | 1 | def add_note(order_id: OrderID, author_id: UserID, text: str) -> None: |
|
156 | """Add a note to the order.""" |
||
157 | order = get_order(order_id) |
||
158 | author = user_service.get_user(author_id) |
||
159 | |||
160 | event_type = 'order-note-added' |
||
161 | data = { |
||
162 | 'author_id': str(author.id), |
||
163 | 'text': text, |
||
164 | } |
||
165 | |||
166 | event_service.create_event(event_type, order.id, data) |
||
167 | |||
168 | |||
169 | 1 | def set_invoiced_flag(order_id: OrderID, initiator_id: UserID) -> None: |
|
170 | """Record that the invoice for that order has been (externally) created.""" |
||
171 | order = _get_order_entity(order_id) |
||
172 | initiator = user_service.get_user(initiator_id) |
||
173 | |||
174 | now = datetime.utcnow() |
||
175 | event_type = 'order-invoiced' |
||
176 | data = { |
||
177 | 'initiator_id': str(initiator.id), |
||
178 | } |
||
179 | |||
180 | event = DbOrderEvent(now, event_type, order.id, data) |
||
181 | db.session.add(event) |
||
182 | |||
183 | order.invoice_created_at = now |
||
184 | |||
185 | db.session.commit() |
||
186 | |||
187 | |||
188 | 1 | def unset_invoiced_flag(order_id: OrderID, initiator_id: UserID) -> None: |
|
189 | """Withdraw record of the invoice for that order having been created.""" |
||
190 | order = _get_order_entity(order_id) |
||
191 | initiator = user_service.get_user(initiator_id) |
||
192 | |||
193 | now = datetime.utcnow() |
||
194 | event_type = 'order-invoiced-withdrawn' |
||
195 | data = { |
||
196 | 'initiator_id': str(initiator.id), |
||
197 | } |
||
198 | |||
199 | event = DbOrderEvent(now, event_type, order.id, data) |
||
200 | db.session.add(event) |
||
201 | |||
202 | order.invoice_created_at = None |
||
203 | |||
204 | db.session.commit() |
||
205 | |||
206 | |||
207 | 1 | View Code Duplication | def set_shipped_flag(order_id: OrderID, initiator_id: UserID) -> None: |
208 | """Mark the order as shipped.""" |
||
209 | order = _get_order_entity(order_id) |
||
210 | initiator = user_service.get_user(initiator_id) |
||
211 | |||
212 | if not order.shipping_required: |
||
213 | raise ValueError('Order contains no items that require shipping.') |
||
214 | |||
215 | now = datetime.utcnow() |
||
216 | event_type = 'order-shipped' |
||
217 | data = { |
||
218 | 'initiator_id': str(initiator.id), |
||
219 | } |
||
220 | |||
221 | event = DbOrderEvent(now, event_type, order.id, data) |
||
222 | db.session.add(event) |
||
223 | |||
224 | order.shipped_at = now |
||
225 | |||
226 | db.session.commit() |
||
227 | |||
228 | |||
229 | 1 | View Code Duplication | def unset_shipped_flag(order_id: OrderID, initiator_id: UserID) -> None: |
230 | """Mark the order as not shipped.""" |
||
231 | order = _get_order_entity(order_id) |
||
232 | initiator = user_service.get_user(initiator_id) |
||
233 | |||
234 | if not order.shipping_required: |
||
235 | raise ValueError('Order contains no items that require shipping.') |
||
236 | |||
237 | now = datetime.utcnow() |
||
238 | event_type = 'order-shipped-withdrawn' |
||
239 | data = { |
||
240 | 'initiator_id': str(initiator.id), |
||
241 | } |
||
242 | |||
243 | event = DbOrderEvent(now, event_type, order.id, data) |
||
244 | db.session.add(event) |
||
245 | |||
246 | order.shipped_at = None |
||
247 | |||
248 | db.session.commit() |
||
249 | |||
250 | |||
251 | 1 | class OrderAlreadyCanceled(Exception): |
|
252 | 1 | pass |
|
253 | |||
254 | |||
255 | 1 | class OrderAlreadyMarkedAsPaid(Exception): |
|
256 | 1 | pass |
|
257 | |||
258 | |||
259 | 1 | def cancel_order( |
|
260 | order_id: OrderID, initiator_id: UserID, reason: str |
||
261 | ) -> ShopOrderCanceled: |
||
262 | """Cancel the order. |
||
263 | |||
264 | Reserved quantities of articles from that order are made available |
||
265 | again. |
||
266 | """ |
||
267 | 1 | order = _get_order_entity(order_id) |
|
268 | |||
269 | 1 | if _is_canceled(order): |
|
270 | raise OrderAlreadyCanceled() |
||
271 | |||
272 | 1 | initiator = user_service.get_user(initiator_id) |
|
273 | 1 | orderer_user = user_service.get_user(order.placed_by_id) |
|
274 | |||
275 | 1 | has_order_been_paid = _is_paid(order) |
|
276 | |||
277 | 1 | now = datetime.utcnow() |
|
278 | |||
279 | 1 | updated_at = now |
|
280 | 1 | payment_state_from = order.payment_state |
|
281 | 1 | payment_state_to = ( |
|
282 | PaymentState.canceled_after_paid |
||
283 | if has_order_been_paid |
||
284 | else PaymentState.canceled_before_paid |
||
285 | ) |
||
286 | |||
287 | 1 | _update_payment_state(order, payment_state_to, updated_at, initiator.id) |
|
288 | 1 | order.cancelation_reason = reason |
|
289 | |||
290 | 1 | event_type = ( |
|
291 | 'order-canceled-after-paid' |
||
292 | if has_order_been_paid |
||
293 | else 'order-canceled-before-paid' |
||
294 | ) |
||
295 | 1 | data = { |
|
296 | 'initiator_id': str(initiator.id), |
||
297 | 'former_payment_state': payment_state_from.name, |
||
298 | 'reason': reason, |
||
299 | } |
||
300 | |||
301 | 1 | event = DbOrderEvent(now, event_type, order.id, data) |
|
302 | 1 | db.session.add(event) |
|
303 | |||
304 | # Make the reserved quantity of articles available again. |
||
305 | 1 | for item in order.items: |
|
306 | 1 | article_service.increase_quantity( |
|
307 | item.article.id, item.quantity, commit=False |
||
308 | ) |
||
309 | |||
310 | 1 | db.session.commit() |
|
311 | |||
312 | 1 | action_service.execute_actions( |
|
313 | _order_to_transfer_object(order), payment_state_to, initiator.id |
||
314 | ) |
||
315 | |||
316 | 1 | return ShopOrderCanceled( |
|
317 | occurred_at=updated_at, |
||
318 | initiator_id=initiator.id, |
||
319 | initiator_screen_name=initiator.screen_name, |
||
320 | order_id=order.id, |
||
321 | order_number=order.order_number, |
||
322 | orderer_id=orderer_user.id, |
||
323 | orderer_screen_name=orderer_user.screen_name, |
||
324 | ) |
||
325 | |||
326 | |||
327 | 1 | def mark_order_as_paid( |
|
328 | order_id: OrderID, |
||
329 | payment_method: PaymentMethod, |
||
330 | initiator_id: UserID, |
||
331 | *, |
||
332 | additional_event_data: Optional[Mapping[str, str]] = None, |
||
333 | ) -> ShopOrderPaid: |
||
334 | """Mark the order as paid.""" |
||
335 | 1 | order = _get_order_entity(order_id) |
|
336 | |||
337 | 1 | if _is_paid(order): |
|
338 | raise OrderAlreadyMarkedAsPaid() |
||
339 | |||
340 | 1 | initiator = user_service.get_user(initiator_id) |
|
341 | 1 | orderer_user = user_service.get_user(order.placed_by_id) |
|
342 | |||
343 | 1 | now = datetime.utcnow() |
|
344 | |||
345 | 1 | updated_at = now |
|
346 | 1 | payment_state_from = order.payment_state |
|
347 | 1 | payment_state_to = PaymentState.paid |
|
348 | |||
349 | 1 | order.payment_method = payment_method |
|
350 | 1 | _update_payment_state(order, payment_state_to, updated_at, initiator.id) |
|
351 | |||
352 | 1 | event_type = 'order-paid' |
|
353 | # Add required, internally set properties after given additional |
||
354 | # ones to ensure the former are not overridden by the latter. |
||
355 | 1 | event_data: OrderEventData = {} |
|
356 | 1 | if additional_event_data is not None: |
|
357 | 1 | event_data.update(additional_event_data) |
|
358 | 1 | event_data.update( |
|
359 | { |
||
360 | 'initiator_id': str(initiator.id), |
||
361 | 'former_payment_state': payment_state_from.name, |
||
362 | 'payment_method': payment_method.name, |
||
363 | } |
||
364 | ) |
||
365 | |||
366 | 1 | event = DbOrderEvent(now, event_type, order.id, event_data) |
|
367 | 1 | db.session.add(event) |
|
368 | |||
369 | 1 | db.session.commit() |
|
370 | |||
371 | 1 | action_service.execute_actions( |
|
372 | _order_to_transfer_object(order), payment_state_to, initiator.id |
||
373 | ) |
||
374 | |||
375 | 1 | return ShopOrderPaid( |
|
376 | occurred_at=updated_at, |
||
377 | initiator_id=initiator.id, |
||
378 | initiator_screen_name=initiator.screen_name, |
||
379 | order_id=order.id, |
||
380 | order_number=order.order_number, |
||
381 | orderer_id=orderer_user.id, |
||
382 | orderer_screen_name=orderer_user.screen_name, |
||
383 | payment_method=payment_method, |
||
384 | ) |
||
385 | |||
386 | |||
387 | 1 | def _update_payment_state( |
|
388 | order: DbOrder, |
||
389 | state: PaymentState, |
||
390 | updated_at: datetime, |
||
391 | initiator_id: UserID, |
||
392 | ) -> None: |
||
393 | 1 | order.payment_state = state |
|
394 | 1 | order.payment_state_updated_at = updated_at |
|
395 | 1 | order.payment_state_updated_by_id = initiator_id |
|
396 | |||
397 | |||
398 | 1 | def delete_order(order_id: OrderID) -> None: |
|
399 | """Delete an order.""" |
||
400 | 1 | order = get_order(order_id) |
|
401 | |||
402 | 1 | db.session.query(DbOrderEvent) \ |
|
403 | .filter_by(order_id=order.id) \ |
||
404 | .delete() |
||
405 | |||
406 | 1 | db.session.query(DbOrderItem) \ |
|
407 | .filter_by(order_number=order.order_number) \ |
||
408 | .delete() |
||
409 | |||
410 | 1 | db.session.query(DbOrder) \ |
|
411 | .filter_by(id=order.id) \ |
||
412 | .delete() |
||
413 | |||
414 | 1 | db.session.commit() |
|
415 | |||
416 | |||
417 | 1 | def count_open_orders(shop_id: ShopID) -> int: |
|
418 | """Return the number of open orders for the shop.""" |
||
419 | return DbOrder.query \ |
||
420 | .for_shop(shop_id) \ |
||
421 | .filter_by(_payment_state=PaymentState.open.name) \ |
||
422 | .count() |
||
423 | |||
424 | |||
425 | 1 | def count_orders_per_payment_state(shop_id: ShopID) -> dict[PaymentState, int]: |
|
426 | """Count orders for the shop, grouped by payment state.""" |
||
427 | counts_by_payment_state = dict.fromkeys(PaymentState, 0) |
||
428 | |||
429 | rows = db.session \ |
||
430 | .query( |
||
431 | DbOrder._payment_state, |
||
432 | db.func.count(DbOrder.id) |
||
433 | ) \ |
||
434 | .filter(DbOrder.shop_id == shop_id) \ |
||
435 | .group_by(DbOrder._payment_state) \ |
||
436 | .all() |
||
437 | |||
438 | for payment_state_str, count in rows: |
||
439 | payment_state = PaymentState[payment_state_str] |
||
440 | counts_by_payment_state[payment_state] = count |
||
441 | |||
442 | return counts_by_payment_state |
||
443 | |||
444 | |||
445 | 1 | def _find_order_entity(order_id: OrderID) -> Optional[DbOrder]: |
|
446 | """Return the order database entity with that id, or `None` if not |
||
447 | found. |
||
448 | """ |
||
449 | 1 | return db.session.query(DbOrder).get(order_id) |
|
450 | |||
451 | |||
452 | 1 | def _get_order_entity(order_id: OrderID) -> DbOrder: |
|
453 | """Return the order database entity with that id, or raise an |
||
454 | exception. |
||
455 | """ |
||
456 | 1 | order = _find_order_entity(order_id) |
|
457 | |||
458 | 1 | if order is None: |
|
459 | raise ValueError(f'Unknown order ID "{order_id}"') |
||
460 | |||
461 | 1 | return order |
|
462 | |||
463 | |||
464 | 1 | def find_order(order_id: OrderID) -> Optional[Order]: |
|
465 | """Return the order with that id, or `None` if not found.""" |
||
466 | 1 | order = _find_order_entity(order_id) |
|
467 | |||
468 | 1 | if order is None: |
|
469 | return None |
||
470 | |||
471 | 1 | return _order_to_transfer_object(order) |
|
472 | |||
473 | |||
474 | 1 | def get_order(order_id: OrderID) -> Order: |
|
475 | """Return the order with that id, or raise an exception.""" |
||
476 | 1 | order = _get_order_entity(order_id) |
|
477 | 1 | return _order_to_transfer_object(order) |
|
478 | |||
479 | |||
480 | 1 | def find_order_with_details(order_id: OrderID) -> Optional[Order]: |
|
481 | """Return the order with that id, or `None` if not found.""" |
||
482 | 1 | order = db.session.query(DbOrder) \ |
|
483 | .options( |
||
484 | db.joinedload(DbOrder.items), |
||
485 | ) \ |
||
486 | .get(order_id) |
||
487 | |||
488 | 1 | if order is None: |
|
489 | 1 | return None |
|
490 | |||
491 | 1 | return _order_to_transfer_object(order) |
|
492 | |||
493 | |||
494 | 1 | def find_order_by_order_number(order_number: OrderNumber) -> Optional[Order]: |
|
495 | """Return the order with that order number, or `None` if not found.""" |
||
496 | order = DbOrder.query \ |
||
497 | .filter_by(order_number=order_number) \ |
||
498 | .one_or_none() |
||
499 | |||
500 | if order is None: |
||
501 | return None |
||
502 | |||
503 | return _order_to_transfer_object(order) |
||
504 | |||
505 | |||
506 | 1 | def find_orders_by_order_numbers( |
|
507 | order_numbers: set[OrderNumber], |
||
508 | ) -> Sequence[Order]: |
||
509 | """Return the orders with those order numbers.""" |
||
510 | if not order_numbers: |
||
511 | return [] |
||
512 | |||
513 | orders = DbOrder.query \ |
||
514 | .filter(DbOrder.order_number.in_(order_numbers)) \ |
||
515 | .all() |
||
516 | |||
517 | return list(map(_order_to_transfer_object, orders)) |
||
518 | |||
519 | |||
520 | 1 | def get_order_count_by_shop_id() -> dict[ShopID, int]: |
|
521 | """Return order count (including 0) per shop, indexed by shop ID.""" |
||
522 | shop_ids_and_order_counts = db.session \ |
||
523 | .query( |
||
524 | DbShop.id, |
||
525 | db.func.count(DbOrder.shop_id) |
||
526 | ) \ |
||
527 | .outerjoin(DbOrder) \ |
||
528 | .group_by(DbShop.id) \ |
||
529 | .all() |
||
530 | |||
531 | return dict(shop_ids_and_order_counts) |
||
532 | |||
533 | |||
534 | 1 | def get_orders_for_shop_paginated( |
|
535 | shop_id: ShopID, |
||
536 | page: int, |
||
537 | per_page: int, |
||
538 | *, |
||
539 | search_term=None, |
||
540 | only_payment_state: Optional[PaymentState] = None, |
||
541 | only_shipped: Optional[bool] = None, |
||
542 | ) -> Pagination: |
||
543 | """Return all orders for that shop, ordered by creation date. |
||
544 | |||
545 | If a payment state is specified, only orders in that state are |
||
546 | returned. |
||
547 | """ |
||
548 | query = DbOrder.query \ |
||
549 | .for_shop(shop_id) \ |
||
550 | .order_by(DbOrder.created_at.desc()) |
||
551 | |||
552 | if search_term: |
||
553 | ilike_pattern = f'%{search_term}%' |
||
554 | query = query \ |
||
555 | .filter(DbOrder.order_number.ilike(ilike_pattern)) |
||
556 | |||
557 | if only_payment_state is not None: |
||
558 | query = query.filter_by(_payment_state=only_payment_state.name) |
||
559 | |||
560 | if only_shipped is not None: |
||
561 | query = query.filter(DbOrder.shipping_required == True) |
||
562 | |||
563 | if only_shipped: |
||
564 | query = query.filter(DbOrder.shipped_at != None) |
||
565 | else: |
||
566 | query = query.filter(DbOrder.shipped_at == None) |
||
567 | |||
568 | return paginate( |
||
569 | query, |
||
570 | page, |
||
571 | per_page, |
||
572 | item_mapper=lambda order: _order_to_transfer_object(order), |
||
573 | ) |
||
574 | |||
575 | |||
576 | 1 | def get_orders_placed_by_user(user_id: UserID) -> Sequence[Order]: |
|
577 | """Return orders placed by the user.""" |
||
578 | 1 | orders = DbOrder.query \ |
|
579 | .options( |
||
580 | db.joinedload(DbOrder.items), |
||
581 | ) \ |
||
582 | .placed_by(user_id) \ |
||
583 | .order_by(DbOrder.created_at.desc()) \ |
||
584 | .all() |
||
585 | |||
586 | 1 | return list(map(_order_to_transfer_object, orders)) |
|
587 | |||
588 | |||
589 | 1 | def get_orders_placed_by_user_for_shop( |
|
590 | user_id: UserID, shop_id: ShopID |
||
591 | ) -> Sequence[Order]: |
||
592 | """Return orders placed by the user in that shop.""" |
||
593 | 1 | orders = DbOrder.query \ |
|
594 | .options( |
||
595 | db.joinedload(DbOrder.items), |
||
596 | ) \ |
||
597 | .for_shop(shop_id) \ |
||
598 | .placed_by(user_id) \ |
||
599 | .order_by(DbOrder.created_at.desc()) \ |
||
600 | .all() |
||
601 | |||
602 | 1 | return list(map(_order_to_transfer_object, orders)) |
|
603 | |||
604 | |||
605 | 1 | def has_user_placed_orders(user_id: UserID, shop_id: ShopID) -> bool: |
|
606 | """Return `True` if the user has placed orders in that shop.""" |
||
607 | 1 | orders_total = DbOrder.query \ |
|
608 | .for_shop(shop_id) \ |
||
609 | .placed_by(user_id) \ |
||
610 | .count() |
||
611 | |||
612 | 1 | return orders_total > 0 |
|
613 | |||
614 | |||
615 | 1 | _PAYMENT_METHOD_LABELS = { |
|
616 | PaymentMethod.bank_transfer: lazy_gettext('bank transfer'), |
||
617 | PaymentMethod.cash: lazy_gettext('cash'), |
||
618 | PaymentMethod.direct_debit: lazy_gettext('direct debit'), |
||
619 | PaymentMethod.free: lazy_gettext('free'), |
||
620 | } |
||
621 | |||
622 | |||
623 | 1 | def find_payment_method_label(payment_method: PaymentMethod) -> Optional[str]: |
|
624 | """Return a label for the payment method.""" |
||
625 | 1 | return _PAYMENT_METHOD_LABELS.get(payment_method) |
|
626 | |||
627 | |||
628 | 1 | def get_payment_date(order_id: OrderID) -> Optional[datetime]: |
|
629 | """Return the date the order has been marked as paid, or `None` if |
||
630 | it has not been paid. |
||
631 | """ |
||
632 | 1 | return db.session \ |
|
633 | .query(DbOrder.payment_state_updated_at) \ |
||
634 | .filter_by(id=order_id) \ |
||
635 | .scalar() |
||
636 | |||
637 | |||
638 | 1 | def _order_to_transfer_object(order: DbOrder) -> Order: |
|
639 | """Create transfer object from order database entity.""" |
||
640 | 1 | address = Address( |
|
641 | country=order.country, |
||
642 | zip_code=order.zip_code, |
||
643 | city=order.city, |
||
644 | street=order.street, |
||
645 | ) |
||
646 | |||
647 | 1 | items = list(map(order_item_to_transfer_object, order.items)) |
|
648 | |||
649 | 1 | state = _get_order_state(order) |
|
650 | 1 | is_open = order.payment_state == PaymentState.open |
|
651 | 1 | is_canceled = _is_canceled(order) |
|
652 | 1 | is_paid = _is_paid(order) |
|
653 | 1 | is_invoiced = order.invoice_created_at is not None |
|
654 | 1 | is_shipping_required = order.shipping_required |
|
655 | 1 | is_shipped = order.shipped_at is not None |
|
656 | |||
657 | 1 | return Order( |
|
658 | id=order.id, |
||
659 | shop_id=order.shop_id, |
||
660 | order_number=order.order_number, |
||
661 | created_at=order.created_at, |
||
662 | placed_by_id=order.placed_by_id, |
||
663 | first_names=order.first_names, |
||
664 | last_name=order.last_name, |
||
665 | address=address, |
||
666 | total_amount=order.total_amount, |
||
667 | items=items, |
||
668 | payment_method=order.payment_method, |
||
669 | payment_state=order.payment_state, |
||
670 | state=state, |
||
671 | is_open=is_open, |
||
672 | is_canceled=is_canceled, |
||
673 | is_paid=is_paid, |
||
674 | is_invoiced=is_invoiced, |
||
675 | is_shipping_required=is_shipping_required, |
||
676 | is_shipped=is_shipped, |
||
677 | cancelation_reason=order.cancelation_reason, |
||
678 | ) |
||
679 | |||
680 | |||
681 | 1 | def order_item_to_transfer_object( |
|
682 | item: DbOrderItem, |
||
683 | ) -> OrderItem: |
||
684 | """Create transfer object from order item database entity.""" |
||
685 | 1 | return OrderItem( |
|
686 | order_number=item.order_number, |
||
687 | article_number=item.article_number, |
||
688 | description=item.description, |
||
689 | unit_price=item.unit_price, |
||
690 | tax_rate=item.tax_rate, |
||
691 | quantity=item.quantity, |
||
692 | line_amount=item.line_amount, |
||
693 | ) |
||
694 | |||
695 | |||
696 | 1 | def _get_order_state(order: DbOrder) -> OrderState: |
|
697 | 1 | is_canceled = _is_canceled(order) |
|
698 | 1 | is_paid = _is_paid(order) |
|
699 | 1 | is_shipping_required = order.shipping_required |
|
700 | 1 | is_shipped = order.shipped_at is not None |
|
701 | |||
702 | 1 | if is_canceled: |
|
703 | 1 | return OrderState.canceled |
|
704 | |||
705 | 1 | if is_paid: |
|
706 | 1 | if not is_shipping_required or is_shipped: |
|
707 | 1 | return OrderState.complete |
|
708 | |||
709 | 1 | return OrderState.open |
|
710 | |||
711 | |||
712 | 1 | def _is_canceled(order: DbOrder) -> bool: |
|
713 | 1 | return order.payment_state in { |
|
714 | PaymentState.canceled_before_paid, |
||
715 | PaymentState.canceled_after_paid, |
||
716 | } |
||
717 | |||
718 | |||
719 | 1 | def _is_paid(order: DbOrder) -> bool: |
|
720 | return order.payment_state == PaymentState.paid |
||
721 |