~homeworkprod/byceps

ref: 4237b3ec9496efe95dcce82bea3207ab9de4d520 byceps/byceps/services/shop/shipping/service.py -rw-r--r-- 3.7 KiB
4237b3ec — Jochen Kupperschmidt Move ticketing blueprint into `site` subpackage 2 years 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
124
125
"""
byceps.services.shop.shipping.service
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

:Copyright: 2006-2020 Jochen Kupperschmidt
:License: Modified BSD, see LICENSE for details.
"""

from collections import Counter, defaultdict
from dataclasses import dataclass
from typing import Dict, Iterator, Sequence, Set

from ..article.models.article import Article as DbArticle

from ....database import db

from ..article.transfer.models import ArticleNumber
from ..order.models.order import Order as DbOrder
from ..order.models.order_item import OrderItem as DbOrderItem
from ..order.transfer.models import PaymentState
from ..shop.transfer.models import ShopID

from .transfer.models import ArticleToShip


def get_articles_to_ship(shop_id: ShopID) -> Sequence[ArticleToShip]:
    """Return articles that need, or likely need, to be shipped soon."""
    relevant_order_payment_states = {
        PaymentState.open,
        PaymentState.paid,
    }

    order_item_quantities = list(
        _find_order_items(shop_id, relevant_order_payment_states)
    )

    article_numbers = {item.article_number for item in order_item_quantities}
    article_descriptions = _get_article_descriptions(article_numbers)

    articles_to_ship = list(
        _aggregate_ordered_article_quantites(
            order_item_quantities, article_descriptions
        )
    )

    articles_to_ship.sort(key=lambda a: a.article_number)

    return articles_to_ship


@dataclass(frozen=True)
class OrderItemQuantity:
    article_number: ArticleNumber
    payment_state: PaymentState
    quantity: int


def _find_order_items(
    shop_id: ShopID, payment_states: Set[PaymentState]
) -> Iterator[OrderItemQuantity]:
    """Return article quantities for the given payment states."""
    payment_state_names = {ps.name for ps in payment_states}

    common_query = DbOrderItem.query \
        .join(DbOrder) \
        .filter(DbOrder.shop_id == shop_id) \
        .options(db.joinedload('order')) \
        .filter(DbOrderItem.shipping_required == True)

    definitive_order_items = common_query \
        .filter(DbOrder._payment_state == PaymentState.paid.name) \
        .filter(DbOrder.shipped_at == None) \
        .all()

    potential_order_items = common_query \
        .filter(DbOrder._payment_state == PaymentState.open.name) \
        .all()

    order_items = definitive_order_items + potential_order_items

    for item in order_items:
        yield OrderItemQuantity(
            item.article_number,
            item.order.payment_state,
            item.quantity
        )


def _aggregate_ordered_article_quantites(
    order_item_quantities: Sequence[OrderItemQuantity],
    article_descriptions: Dict[ArticleNumber, str],
) -> Iterator[ArticleToShip]:
    """Aggregate article quantities per payment state."""
    d = defaultdict(Counter)

    for item in order_item_quantities:
        d[item.article_number][item.payment_state] += item.quantity

    for article_number, counter in d.items():
        description = article_descriptions[article_number]
        quantity_paid = counter[PaymentState.paid]
        quantity_open = counter[PaymentState.open]

        yield ArticleToShip(
            article_number,
            description,
            quantity_paid,
            quantity_open,
            quantity_total=quantity_paid + quantity_open,
        )


def _get_article_descriptions(
    article_numbers: Set[ArticleNumber],
) -> Dict[ArticleNumber, str]:
    """Look up description texts of the specified articles."""
    if not article_numbers:
        return []

    articles = DbArticle.query \
        .options(db.load_only('item_number', 'description')) \
        .filter(DbArticle.item_number.in_(article_numbers)) \
        .all()

    return {a.item_number: a.description for a in articles}