byceps.services.shop.article.service   A
last analyzed

Complexity

Total Complexity 35

Size/Duplication

Total Lines 443
Duplicated Lines 0 %

Test Coverage

Coverage 64.39%

Importance

Changes 0
Metric Value
eloc 271
dl 0
loc 443
ccs 85
cts 132
cp 0.6439
rs 9.6
c 0
b 0
f 0
wmc 35

23 Functions

Rating   Name   Duplication   Size   Complexity  
A find_db_article() 0 5 1
A unattach_article() 0 7 1
A get_article() 0 11 2
A decrease_quantity() 0 10 2
A delete_article() 0 7 1
A update_article() 0 28 1
A increase_quantity() 0 10 2
A find_attached_article() 0 5 1
A create_article() 0 28 1
A find_article() 0 8 2
A attach_article() 0 12 1
A find_article_with_details() 0 10 1
A _get_db_article() 0 11 2
A get_attachable_articles() 0 20 1
B sum_ordered_articles_by_payment_state() 0 60 5
A get_article_compilation_for_single_article() 0 19 1
A _add_attached_articles() 0 10 2
A get_articles_for_shop() 0 9 1
A get_articles_for_shop_paginated() 0 12 1
A _db_entity_to_article() 0 17 1
A is_article_available_now() 0 8 1
A get_articles_by_numbers() 0 13 2
A get_article_compilation_for_orderable_articles() 0 39 2
1
"""
2
byceps.services.shop.article.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 decimal import Decimal
12 1
from typing import Optional, Sequence
13
14 1
from ....database import db, Pagination
15
16 1
from ..order.dbmodels.line_item import LineItem as DbLineItem
17 1
from ..order.dbmodels.order import Order as DbOrder
18 1
from ..order.transfer.models import PaymentState
19 1
from ..shop.dbmodels import Shop as DbShop
20 1
from ..shop.transfer.models import ShopID
21
22 1
from .dbmodels.article import Article as DbArticle
23 1
from .dbmodels.attached_article import AttachedArticle as DbAttachedArticle
24 1
from .models.compilation import ArticleCompilation, ArticleCompilationItem
25 1
from .transfer.models import (
26
    Article,
27
    ArticleID,
28
    ArticleNumber,
29
    ArticleType,
30
    AttachedArticleID,
31
)
32
33
34 1
class UnknownArticleId(ValueError):
35 1
    pass
36
37
38 1
def create_article(
39
    shop_id: ShopID,
40
    item_number: ArticleNumber,
41
    type_: ArticleType,
42
    description: str,
43
    price: Decimal,
44
    tax_rate: Decimal,
45
    total_quantity: int,
46
    max_quantity_per_order: int,
47
    processing_required: bool,
48
) -> Article:
49
    """Create an article."""
50 1
    article = DbArticle(
51
        shop_id,
52
        item_number,
53
        type_,
54
        description,
55
        price,
56
        tax_rate,
57
        total_quantity,
58
        max_quantity_per_order,
59
        processing_required,
60
    )
61
62 1
    db.session.add(article)
63 1
    db.session.commit()
64
65 1
    return _db_entity_to_article(article)
66
67
68 1
def update_article(
69
    article_id: ArticleID,
70
    description: str,
71
    price: Decimal,
72
    tax_rate: Decimal,
73
    available_from: Optional[datetime],
74
    available_until: Optional[datetime],
75
    total_quantity: int,
76
    max_quantity_per_order: int,
77
    not_directly_orderable: bool,
78
    separate_order_required: bool,
79
) -> Article:
80
    """Update the article."""
81
    article = _get_db_article(article_id)
82
83
    article.description = description
84
    article.price = price
85
    article.tax_rate = tax_rate
86
    article.available_from = available_from
87
    article.available_until = available_until
88
    article.total_quantity = total_quantity
89
    article.max_quantity_per_order = max_quantity_per_order
90
    article.not_directly_orderable = not_directly_orderable
91
    article.separate_order_required = separate_order_required
92
93
    db.session.commit()
94
95
    return _db_entity_to_article(article)
96
97
98 1
def attach_article(
99
    article_number_to_attach: ArticleNumber,
100
    quantity: int,
101
    article_number_to_attach_to: ArticleNumber,
102
) -> None:
103
    """Attach an article to another article."""
104
    attached_article = DbAttachedArticle(
105
        article_number_to_attach, quantity, article_number_to_attach_to
106
    )
107
108
    db.session.add(attached_article)
109
    db.session.commit()
110
111
112 1
def unattach_article(attached_article_id: AttachedArticleID) -> None:
113
    """Unattach an article from another."""
114
    db.session.query(DbAttachedArticle) \
115
        .filter_by(id=attached_article_id) \
116
        .delete()
117
118
    db.session.commit()
119
120
121 1
def increase_quantity(
122
    article_id: ArticleID, quantity_to_increase_by: int, *, commit: bool = True
123
) -> None:
124
    """Increase article quantity by the given value."""
125 1
    db.session.query(DbArticle) \
126
        .filter_by(id=article_id) \
127
        .update({'quantity': DbArticle.quantity + quantity_to_increase_by})
128
129 1
    if commit:
130
        db.session.commit()
131
132
133 1
def decrease_quantity(
134
    article_id: ArticleID, quantity_to_decrease_by: int, *, commit: bool = True
135
) -> None:
136
    """Decrease article quantity by the given value."""
137 1
    db.session.query(DbArticle) \
138
        .filter_by(id=article_id) \
139
        .update({'quantity': DbArticle.quantity - quantity_to_decrease_by})
140
141 1
    if commit:
142
        db.session.commit()
143
144
145 1
def delete_article(article_id: ArticleID) -> None:
146
    """Delete an article."""
147 1
    db.session.query(DbArticle) \
148
        .filter_by(id=article_id) \
149
        .delete()
150
151 1
    db.session.commit()
152
153
154 1
def find_article(article_id: ArticleID) -> Optional[Article]:
155
    """Return the article with that ID, or `None` if not found."""
156 1
    article = find_db_article(article_id)
157
158 1
    if article is None:
159
        return None
160
161 1
    return _db_entity_to_article(article)
162
163
164 1
def get_article(article_id: ArticleID) -> Article:
165
    """Return the article with that ID.
166
167
    Raise an exception if not found.
168
    """
169 1
    article = find_article(article_id)
170
171 1
    if article is None:
172
        raise UnknownArticleId(article_id)
173
174 1
    return article
175
176
177 1
def find_db_article(article_id: ArticleID) -> Optional[DbArticle]:
178
    """Return the database entity for the article with that ID, or
179
    `None` if not found.
180
    """
181 1
    return db.session.query(DbArticle).get(article_id)
182
183
184 1
def _get_db_article(article_id: ArticleID) -> DbArticle:
185
    """Return the database entity for the article with that id.
186
187
    Raise an exception if not found.
188
    """
189 1
    article = find_db_article(article_id)
190
191 1
    if article is None:
192
        raise UnknownArticleId(article_id)
193
194 1
    return article
195
196
197 1
def find_article_with_details(article_id: ArticleID) -> Optional[DbArticle]:
198
    """Return the article with that ID, or `None` if not found."""
199
    return db.session.query(DbArticle) \
200
        .options(
201
            db.joinedload(DbArticle.articles_attached_to)
202
                .joinedload(DbAttachedArticle.article),
203
            db.joinedload(DbArticle.attached_articles)
204
                .joinedload(DbAttachedArticle.article),
205
        ) \
206
        .get(article_id)
207
208
209 1
def find_attached_article(
210
    attached_article_id: AttachedArticleID,
211
) -> Optional[DbAttachedArticle]:
212
    """Return the attached article with that ID, or `None` if not found."""
213
    return db.session.query(DbAttachedArticle).get(attached_article_id)
214
215
216 1
def get_articles_by_numbers(
217
    article_numbers: set[ArticleNumber],
218
) -> set[Article]:
219
    """Return the articles with those numbers."""
220
    if not article_numbers:
221
        return set()
222
223
    rows = db.session \
224
        .query(DbArticle) \
225
        .filter(DbArticle.item_number.in_(article_numbers)) \
226
        .all()
227
228
    return {_db_entity_to_article(row) for row in rows}
229
230
231 1
def get_articles_for_shop(shop_id: ShopID) -> Sequence[Article]:
232
    """Return all articles for that shop, ordered by article number."""
233
    rows = db.session \
234
        .query(DbArticle) \
235
        .filter_by(shop_id=shop_id) \
236
        .order_by(DbArticle.item_number) \
237
        .all()
238
239
    return [_db_entity_to_article(row) for row in rows]
240
241
242 1
def get_articles_for_shop_paginated(
243
    shop_id: ShopID, page: int, per_page: int
244
) -> Pagination:
245
    """Return all articles for that shop, paginated.
246
247
    Ordered by article number, reversed.
248
    """
249
    return db.session \
250
        .query(DbArticle) \
251
        .filter_by(shop_id=shop_id) \
252
        .order_by(DbArticle.item_number.desc()) \
253
        .paginate(page, per_page)
254
255
256 1
def get_article_compilation_for_orderable_articles(
257
    shop_id: ShopID,
258
) -> ArticleCompilation:
259
    """Return a compilation of the articles which can be ordered from
260
    that shop, less the ones that are only orderable in a dedicated
261
    order.
262
    """
263 1
    now = datetime.utcnow()
264
265 1
    orderable_articles = (db.session
266
        .query(DbArticle)
267
        .filter_by(shop_id=shop_id)
268
        .filter_by(not_directly_orderable=False)
269
        .filter_by(separate_order_required=False)
270
271
        # Select only articles that are available in between the
272
        # temporal boundaries for this article, if specified.
273
        .filter(db.or_(
274
            DbArticle.available_from == None,
275
            now >= DbArticle.available_from
276
        ))
277
        .filter(db.or_(
278
            DbArticle.available_until == None,
279
            now < DbArticle.available_until
280
        ))
281
282
        .order_by(DbArticle.description)
283
        .all())
284
285 1
    compilation = ArticleCompilation()
286
287 1
    for article in orderable_articles:
288 1
        compilation.append(
289
            ArticleCompilationItem(_db_entity_to_article(article))
290
        )
291
292 1
        _add_attached_articles(compilation, article.attached_articles)
293
294 1
    return compilation
295
296
297 1
def get_article_compilation_for_single_article(
298
    article_id: ArticleID, *, fixed_quantity: Optional[int] = None
299
) -> ArticleCompilation:
300
    """Return a compilation built from just the given article plus the
301
    articles attached to it (if any).
302
    """
303 1
    article = _get_db_article(article_id)
304
305 1
    compilation = ArticleCompilation()
306
307 1
    compilation.append(
308
        ArticleCompilationItem(
309
            _db_entity_to_article(article), fixed_quantity=fixed_quantity
310
        )
311
    )
312
313 1
    _add_attached_articles(compilation, article.attached_articles)
314
315 1
    return compilation
316
317
318 1
def _add_attached_articles(
319
    compilation: ArticleCompilation,
320
    attached_articles: Sequence[DbAttachedArticle],
321
) -> None:
322
    """Add the attached articles to the compilation."""
323 1
    for attached_article in attached_articles:
324
        compilation.append(
325
            ArticleCompilationItem(
326
                _db_entity_to_article(attached_article.article),
327
                fixed_quantity=attached_article.quantity,
328
            )
329
        )
330
331
332 1
def get_attachable_articles(article_id: ArticleID) -> set[Article]:
333
    """Return the articles that can be attached to that article."""
334
    article = _get_db_article(article_id)
335
336
    attached_articles = {
337
        attached.article for attached in article.attached_articles
338
    }
339
340
    unattachable_articles = {article}.union(attached_articles)
341
342
    unattachable_article_ids = {article.id for article in unattachable_articles}
343
344
    rows = db.session \
345
        .query(DbArticle) \
346
        .filter_by(shop_id=article.shop_id) \
347
        .filter(db.not_(DbArticle.id.in_(unattachable_article_ids))) \
348
        .order_by(DbArticle.item_number) \
349
        .all()
350
351
    return {_db_entity_to_article(row) for row in rows}
352
353
354 1
def is_article_available_now(article: Article) -> bool:
355
    """Return `True` if the article is available at this moment in time."""
356 1
    start = article.available_from
357 1
    end = article.available_until
358
359 1
    now = datetime.utcnow()
360
361 1
    return (start is None or start <= now) and (end is None or now < end)
362
363
364 1
def sum_ordered_articles_by_payment_state(
365
    shop_ids: set[ShopID],
366
) -> list[tuple[ShopID, ArticleNumber, str, PaymentState, int]]:
367
    """Sum ordered articles for those shops, grouped by order payment state."""
368 1
    subquery = db.session \
369
        .query(
370
            DbLineItem.article_number,
371
            DbOrder._payment_state.label('payment_state'),
372
            db.func.sum(DbLineItem.quantity).label('quantity')
373
        ) \
374
        .join(DbOrder) \
375
        .group_by(DbLineItem.article_number, DbOrder._payment_state) \
376
        .subquery()
377
378 1
    rows = db.session \
379
        .query(
380
            DbArticle.shop_id,
381
            DbArticle.item_number,
382
            DbArticle.description,
383
            subquery.c.payment_state,
384
            subquery.c.quantity
385
        ) \
386
        .outerjoin(subquery,
387
            db.and_(DbArticle.item_number == subquery.c.article_number)) \
388
        .filter(DbArticle.shop_id.in_(shop_ids)) \
389
        .order_by(DbArticle.item_number, subquery.c.payment_state) \
390
        .all()
391
392 1
    shop_ids_and_article_numbers_and_descriptions = {
393
        (row[0], row[1], row[2]) for row in rows
394
    }  # Remove duplicates.
395
396 1
    quantities = {}
397
398 1
    for (
399
        shop_id,
400
        article_number,
401
        description,
402
        payment_state_name,
403
        quantity,
404
    ) in rows:
405
        if payment_state_name is None:
406
            continue
407
408
        payment_state = PaymentState[payment_state_name]
409
        key = (shop_id, article_number, description, payment_state)
410
411
        quantities[key] = quantity
412
413 1
    def generate():
414 1
        for shop_id, article_number, description in sorted(
415
            shop_ids_and_article_numbers_and_descriptions
416
        ):
417
            for payment_state in PaymentState:
418
                key = (shop_id, article_number, description, payment_state)
419
                quantity = quantities.get(key, 0)
420
421
                yield shop_id, article_number, description, payment_state, quantity
422
423 1
    return list(generate())
424
425
426 1
def _db_entity_to_article(article: DbArticle) -> Article:
427 1
    return Article(
428
        id=article.id,
429
        shop_id=article.shop_id,
430
        item_number=article.item_number,
431
        type_=article.type_,
432
        description=article.description,
433
        price=article.price,
434
        tax_rate=article.tax_rate,
435
        available_from=article.available_from,
436
        available_until=article.available_until,
437
        total_quantity=article.total_quantity,
438
        quantity=article.quantity,
439
        max_quantity_per_order=article.max_quantity_per_order,
440
        not_directly_orderable=article.not_directly_orderable,
441
        separate_order_required=article.separate_order_required,
442
        processing_required=article.processing_required,
443
    )
444