~homeworkprod/byceps

ref: c6b428b85f1f55355008cc2d99dd87b0f6d5bf83 byceps/byceps/services/shop/article/sequence_service.py -rw-r--r-- 3.2 KiB
c6b428b8 — Jochen Kupperschmidt Turn `find_article_number_sequence` into `get_article_number_sequence` 2 months ago
                                                                                
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
"""
byceps.services.shop.article.sequence_service
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

:Copyright: 2006-2021 Jochen Kupperschmidt
:License: Revised BSD (see `LICENSE` file for details)
"""

from __future__ import annotations
from typing import Optional

from sqlalchemy.exc import IntegrityError

from ....database import db

from ..shop.transfer.models import ShopID

from .dbmodels.number_sequence import ArticleNumberSequence as DbArticleNumberSequence
from .transfer.models import (
    ArticleNumber,
    ArticleNumberSequence,
    ArticleNumberSequenceID,
)


def create_article_number_sequence(
    shop_id: ShopID, prefix: str, *, value: Optional[int] = None
) -> Optional[ArticleNumberSequenceID]:
    """Create an article number sequence.

    Return the resulting sequence's ID, or `None` if the sequence could
    not be created.
    """
    sequence = DbArticleNumberSequence(shop_id, prefix, value=value)

    db.session.add(sequence)

    try:
        db.session.commit()
    except IntegrityError as e:
        db.session.rollback()
        return None

    return sequence.id


def delete_article_number_sequence(
    sequence_id: ArticleNumberSequenceID,
) -> None:
    """Delete the article number sequence."""
    db.session.query(DbArticleNumberSequence) \
        .filter_by(id=sequence_id) \
        .delete()

    db.session.commit()


def get_article_number_sequence(
    sequence_id: ArticleNumberSequenceID,
) -> ArticleNumberSequence:
    """Return the article number sequence, or raise an exception."""
    sequence = DbArticleNumberSequence.query \
        .filter_by(id=sequence_id) \
        .one_or_none()

    if sequence is None:
        raise ValueError(f'Unknown article number sequence ID "{sequence_id}"')

    return _db_entity_to_article_number_sequence(sequence)


def find_article_number_sequences_for_shop(
    shop_id: ShopID,
) -> list[ArticleNumberSequence]:
    """Return the article number sequences defined for that shop."""
    sequences = DbArticleNumberSequence.query \
        .filter_by(shop_id=shop_id) \
        .all()

    return [
        _db_entity_to_article_number_sequence(sequence)
        for sequence in sequences
    ]


class ArticleNumberGenerationFailed(Exception):
    """Indicate that generating a prefixed, sequential article number
    has failed.
    """

    def __init__(self, message: str) -> None:
        self.message = message


def generate_article_number(
    sequence_id: ArticleNumberSequenceID,
) -> ArticleNumber:
    """Generate and reserve the next article number from this sequence."""
    sequence = DbArticleNumberSequence.query \
        .filter_by(id=sequence_id) \
        .with_for_update() \
        .one_or_none()

    if sequence is None:
        raise ArticleNumberGenerationFailed(
            f'No article number sequence found for ID "{sequence_id}".'
        )

    sequence.value = DbArticleNumberSequence.value + 1
    db.session.commit()

    return ArticleNumber(f'{sequence.prefix}{sequence.value:05d}')


def _db_entity_to_article_number_sequence(
    sequence: DbArticleNumberSequence,
) -> ArticleNumberSequence:
    return ArticleNumberSequence(
        sequence.id,
        sequence.shop_id,
        sequence.prefix,
        sequence.value,
    )