byceps.services.shop.order.service   F
last analyzed

Complexity

Total Complexity 60

Size/Duplication

Total Lines 721
Duplicated Lines 5.55 %

Test Coverage

Coverage 69.06%

Importance

Changes 0
Metric Value
eloc 458
dl 40
loc 721
rs 3.6
c 0
b 0
f 0
ccs 183
cts 265
cp 0.6906
wmc 60

34 Functions

Rating   Name   Duplication   Size   Complexity  
A get_orders_placed_by_user_for_shop() 0 14 1
A add_note() 0 12 1
A find_order() 0 8 2
A _find_order_entity() 0 5 1
A delete_order() 0 17 1
A _reduce_article_stock() 0 7 2
A set_invoiced_flag() 0 17 1
A find_order_with_details() 0 12 2
A find_order_by_order_number() 0 10 2
A _build_order() 0 18 1
A _order_to_transfer_object() 0 40 1
A _get_order_entity() 0 10 2
A set_shipped_flag() 20 20 2
A unset_shipped_flag() 20 20 2
A find_orders_by_order_numbers() 0 12 2
B place_order() 0 52 2
A find_payment_method_label() 0 3 1
A _get_order_state() 0 14 5
A _update_payment_state() 0 9 1
A unset_invoiced_flag() 0 17 1
A get_order() 0 4 1
A count_orders_per_payment_state() 0 18 2
A _is_canceled() 0 4 1
A get_order_count_by_shop_id() 0 12 1
A _is_paid() 0 2 1
A get_orders_placed_by_user() 0 11 1
B get_orders_for_shop_paginated() 0 39 6
B cancel_order() 0 65 5
B mark_order_as_paid() 0 57 3
A has_user_placed_orders() 0 8 1
A get_payment_date() 0 8 1
A count_open_orders() 0 6 1
A _build_line_items() 0 16 2
A line_item_to_transfer_object() 0 12 1

How to fix   Duplicated Code    Complexity   

Duplicated Code

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:

Complexity

 Tip:   Before tackling complexity, make sure that you eliminate any duplication first. This often can reduce the size of classes significantly.

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.line_item import LineItem as DbLineItem
32 1
from .dbmodels.order import Order as DbOrder
33 1
from .dbmodels.order_event import OrderEvent as DbOrderEvent, OrderEventData
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
    LineItem,
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
    line_items = list(_build_line_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 line_items
77
    )
78
79 1
    db.session.add(order)
80 1
    db.session.add_all(line_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_line_items(cart: Cart, order: DbOrder) -> Iterator[DbLineItem]:
128
    """Build line 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 DbLineItem(
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(DbLineItem) \
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(line_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 line_item_to_transfer_object(
682
    item: DbLineItem,
683
) -> LineItem:
684
    """Create transfer object from line item database entity."""
685 1
    return LineItem(
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