Completed
Push — main ( b7acb4...84e7fd )
by Jochen
03:57
created

create_order_number_sequence()   A

Complexity

Conditions 1

Size

Total Lines 10
Code Lines 6

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 5
CRAP Score 1

Importance

Changes 0
Metric Value
cc 1
eloc 6
nop 4
dl 0
loc 10
ccs 5
cts 5
cp 1
crap 1
rs 10
c 0
b 0
f 0
1
"""
2
byceps.services.shop.order.sequence_service
3
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
4
5
:Copyright: 2006-2020 Jochen Kupperschmidt
6
:License: Modified BSD, see LICENSE for details.
7
"""
8
9 1
from typing import List, Optional
10
11 1
from ....database import db
12
13 1
from ..sequence.models import NumberSequence as DbNumberSequence
14 1
from ..sequence.transfer.models import Purpose
15 1
from ..shop.transfer.models import ShopID
16
17 1
from .transfer.models import (
18
    OrderNumber,
19
    OrderNumberSequence,
20
    OrderNumberSequenceID,
21
)
22
23
24 1
def create_order_number_sequence(
25
    shop_id: ShopID, prefix: str, *, value: Optional[int] = None
26
) -> OrderNumberSequenceID:
27
    """Create an order number sequence."""
28 1
    sequence = DbNumberSequence(shop_id, Purpose.order, prefix, value=value)
29
30 1
    db.session.add(sequence)
31 1
    db.session.commit()
32
33 1
    return sequence.id
34
35
36 1
def delete_order_number_sequence(sequence_id: OrderNumberSequenceID) -> None:
37
    """Delete the order number sequence."""
38 1
    db.session.query(DbNumberSequence) \
39
        .filter_by(id=sequence_id) \
40
        .delete()
41
42 1
    db.session.commit()
43
44
45 1
def find_order_number_sequence(
46
    sequence_id: OrderNumberSequenceID,
47
) -> Optional[OrderNumberSequence]:
48
    """Return the order number sequence, or `None` if the sequence ID
49
    is unknown or if the sequence's purpose is not order numbers.
50
    """
51 1
    sequence = DbNumberSequence.query \
52
        .filter_by(id=sequence_id) \
53
        .filter_by(_purpose=Purpose.order.name) \
54
        .one_or_none()
55
56 1
    if sequence is None:
57
        return None
58
59 1
    return _db_entity_to_order_number_sequence(sequence)
60
61
62 1
def find_order_number_sequences_for_shop(
63
    shop_id: ShopID,
64
) -> List[OrderNumberSequence]:
65
    """Return the order number sequences defined for that shop."""
66
    sequences = DbNumberSequence.query \
67
        .filter_by(shop_id=shop_id) \
68
        .filter_by(_purpose=Purpose.order.name) \
69
        .all()
70
71
    return [
72
        _db_entity_to_order_number_sequence(sequence) for sequence in sequences
73
    ]
74
75
76 1
class OrderNumberGenerationFailed(Exception):
77
    """Indicate that generating a prefixed, sequential order number has
78
    failed.
79
    """
80
81 1
    def __init__(self, message: str) -> None:
82
        self.message = message
83
84
85 1
def generate_order_number(sequence_id: OrderNumberSequenceID) -> OrderNumber:
86
    """Generate and reserve an unused, unique order number from this
87
    sequence.
88
    """
89 1
    sequence = DbNumberSequence.query \
90
        .filter_by(id=sequence_id) \
91
        .filter_by(_purpose=Purpose.order.name) \
92
        .with_for_update() \
93
        .one_or_none()
94
95 1
    if sequence is None:
96
        raise OrderNumberGenerationFailed(
97
            f'No order number sequence found for ID "{sequence_id}".'
98
        )
99
100 1
    sequence.value = DbNumberSequence.value + 1
101 1
    db.session.commit()
102
103 1
    return OrderNumber(f'{sequence.prefix}{sequence.value:05d}')
104
105
106 1
def _db_entity_to_order_number_sequence(
107
    sequence: DbNumberSequence,
108
) -> OrderNumberSequence:
109 1
    return OrderNumberSequence(
110
        sequence.id,
111
        sequence.shop_id,
112
        sequence.prefix,
113
        sequence.value,
114
    )
115