~homeworkprod/byceps

ref: bcef32cd2dc329bcf81057794883008f5c040ac8 byceps/byceps/services/shop/order/action_service.py -rw-r--r-- 3.9 KiB
bcef32cd — Jochen Kupperschmidt Work around Jinja 3.0.0 bug with `for` inside of `set` block 1 year, 7 days 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
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
"""
byceps.services.shop.order.action_service
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

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

from __future__ import annotations
from typing import Callable, Sequence

from ....database import db
from ....typing import UserID

from ..article.transfer.models import ArticleNumber

from .actions.award_badge import award_badge
from .actions.create_ticket_bundles import create_ticket_bundles
from .actions.create_tickets import create_tickets
from .actions.revoke_ticket_bundles import revoke_ticket_bundles
from .actions.revoke_tickets import revoke_tickets
from .dbmodels.order_action import OrderAction, Parameters
from .transfer.models import Order, PaymentState


OrderActionType = Callable[[Order, ArticleNumber, int, Parameters], None]


PROCEDURES_BY_NAME = {
    'award_badge': award_badge,
    'create_ticket_bundles': create_ticket_bundles,
    'revoke_ticket_bundles': revoke_ticket_bundles,
    'create_tickets': create_tickets,
    'revoke_tickets': revoke_tickets,
}


# -------------------------------------------------------------------- #
# creation/removal


def create_action(
    article_number: ArticleNumber,
    payment_state: PaymentState,
    procedure: str,
    parameters: Parameters,
) -> None:
    """Create an order action."""
    action = OrderAction(article_number, payment_state, procedure, parameters)

    db.session.add(action)
    db.session.commit()


def delete_actions(article_number: ArticleNumber) -> None:
    """Delete order actions for an article."""
    db.session.query(OrderAction) \
        .filter_by(article_number=article_number) \
        .delete()

    db.session.commit()


# -------------------------------------------------------------------- #
# retrieval


def get_actions_for_article(
    article_number: ArticleNumber,
) -> Sequence[OrderAction]:
    """Return the order actions defined for that article."""
    return OrderAction.query \
        .filter_by(article_number=article_number) \
        .all()


# -------------------------------------------------------------------- #
# execution


def execute_actions(
    order: Order, payment_state: PaymentState, initiator_id: UserID
) -> None:
    """Execute relevant actions for this order in its new payment state."""
    article_numbers = {item.article_number for item in order.items}

    if not article_numbers:
        return

    quantities_by_article_number = {
        item.article_number: item.quantity for item in order.items
    }

    actions = _get_actions(article_numbers, payment_state)

    for action in actions:
        article_quantity = quantities_by_article_number[action.article_number]

        _execute_procedure(order, action, article_quantity, initiator_id)


def _get_actions(
    article_numbers: set[ArticleNumber], payment_state: PaymentState
) -> Sequence[OrderAction]:
    """Return the order actions for those article numbers."""
    return OrderAction.query \
        .filter(OrderAction.article_number.in_(article_numbers)) \
        .filter_by(_payment_state=payment_state.name) \
        .all()


def _execute_procedure(
    order: Order,
    action: OrderAction,
    article_quantity: int,
    initiator_id: UserID,
) -> None:
    """Execute the procedure configured for that order action."""
    article_number = action.article_number
    procedure_name = action.procedure
    params = action.parameters

    procedure = _get_procedure(procedure_name, article_number)

    procedure(order, article_number, article_quantity, initiator_id, params)


def _get_procedure(name: str, article_number: ArticleNumber) -> OrderActionType:
    """Return procedure with that name, or raise an exception if the
    name is not registerd.
    """
    procedure = PROCEDURES_BY_NAME.get(name)

    if procedure is None:
        raise Exception(
            f"Unknown procedure '{name}' configured "
            f"for article number '{article_number}'."
        )

    return procedure