Passed
Push — main ( 68ee7a...c0bc2e )
by Jochen
04:34
created

_get_additional_data_for_order_note_added()   A

Complexity

Conditions 1

Size

Total Lines 6
Code Lines 5

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 1
CRAP Score 1.4218

Importance

Changes 0
Metric Value
cc 1
eloc 5
nop 1
dl 0
loc 6
ccs 1
cts 4
cp 0.25
crap 1.4218
rs 10
c 0
b 0
f 0
1
"""
2
byceps.blueprints.admin.shop.order.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 dataclasses import dataclass
11 1
import dataclasses
12 1
from typing import Iterator, Sequence
13
14 1
from .....services.shop.article import service as article_service
15 1
from .....services.shop.article.transfer.models import Article, ArticleNumber
16 1
from .....services.shop.order.dbmodels.order_event import (
17
    OrderEvent,
18
    OrderEventData,
19
)
20 1
from .....services.shop.order import event_service as order_event_service
21 1
from .....services.shop.order import service as order_service
22 1
from .....services.shop.order.transfer.models import Order, OrderID
23 1
from .....services.ticketing import category_service as ticket_category_service
24 1
from .....services.user.dbmodels.user import User as DbUser
25 1
from .....services.user import service as user_service
26 1
from .....services.user.transfer.models import User
27 1
from .....services.user_badge import badge_service as user_badge_service
28 1
from .....typing import UserID
29
30
31 1
@dataclass(frozen=True)
32 1
class OrderWithOrderer(Order):
33 1
    placed_by: DbUser
34
35
36 1
def extend_order_tuples_with_orderer(
37
    orders: Sequence[Order],
38
) -> Iterator[OrderWithOrderer]:
39
    orderer_ids = {order.placed_by_id for order in orders}
40
    orderers = user_service.find_users(orderer_ids, include_avatars=True)
41
    orderers_by_id = user_service.index_users_by_id(orderers)
42
43
    for order in orders:
44
        orderer = orderers_by_id[order.placed_by_id]
45
        values = dataclasses.astuple(order) + (orderer,)
46
        yield OrderWithOrderer(*values)
47
48
49 1
def get_articles_by_item_number(order: Order) -> dict[ArticleNumber, Article]:
50
    numbers = {item.article_number for item in order.items}
51
52
    articles = article_service.get_articles_by_numbers(numbers)
53
54
    return {article.item_number: article for article in articles}
55
56
57 1
def get_events(order_id: OrderID) -> Iterator[OrderEventData]:
58
    events = order_event_service.get_events_for_order(order_id)
59
    events.insert(0, _fake_order_placement_event(order_id))
60
61
    user_ids = {
62
        event.data['initiator_id']
63
        for event in events
64
        if 'initiator_id' in event.data
65
    }
66
    users = user_service.find_users(user_ids, include_avatars=True)
67
    users_by_id = {str(user.id): user for user in users}
68
69
    for event in events:
70
        data = {
71
            'event': event.event_type,
72
            'occurred_at': event.occurred_at,
73
            'data': event.data,
74
        }
75
76
        additional_data = _get_additional_data(event, users_by_id)
77
        data.update(additional_data)
78
79
        yield data
80
81
82 1
def _fake_order_placement_event(order_id: OrderID) -> OrderEvent:
83
    order = order_service.find_order_with_details(order_id)
84
    if order is None:
85
        raise ValueError('Unknown order ID')
86
87
    data = {
88
        'initiator_id': str(order.placed_by_id),
89
    }
90
91
    return OrderEvent(order.created_at, 'order-placed', order.id, data)
92
93
94 1
def _get_additional_data(
95
    event: OrderEvent, users_by_id: dict[UserID, User]
96
) -> OrderEventData:
97
    if event.event_type == 'badge-awarded':
98
        return _get_additional_data_for_badge_awarded(event)
99
    elif event.event_type == 'order-note-added':
100
        return _get_additional_data_for_order_note_added(event)
101
    elif event.event_type == 'ticket-bundle-created':
102
        return _get_additional_data_for_ticket_bundle_created(event)
103
    elif event.event_type == 'ticket-bundle-revoked':
104
        return _get_additional_data_for_ticket_bundle_revoked(
105
            event, users_by_id
106
        )
107
    elif event.event_type == 'ticket-created':
108
        return _get_additional_data_for_ticket_created(event)
109
    elif event.event_type == 'ticket-revoked':
110
        return _get_additional_data_for_ticket_revoked(event, users_by_id)
111
    else:
112
        return _get_additional_data_for_standard_event(event, users_by_id)
113
114
115 1
def _get_additional_data_for_standard_event(
116
    event: OrderEvent, users_by_id: dict[UserID, User]
117
) -> OrderEventData:
118
    initiator_id = event.data['initiator_id']
119
120
    return {
121
        'initiator': users_by_id[initiator_id],
122
    }
123
124
125 1
def _get_additional_data_for_badge_awarded(event: OrderEvent) -> OrderEventData:
126
    badge_id = event.data['badge_id']
127
    badge = user_badge_service.get_badge(badge_id)
128
129
    recipient_id = event.data['recipient_id']
130
    recipient = user_service.get_user(recipient_id)
131
132
    return {
133
        'badge_label': badge.label,
134
        'recipient': recipient,
135
    }
136
137
138 1
def _get_additional_data_for_order_note_added(event: OrderEvent) -> OrderEventData:
139
    author_id = event.data['author_id']
140
    author = user_service.get_user(author_id)
141
142
    return {
143
        'author': author,
144
    }
145
146
147 1
def _get_additional_data_for_ticket_bundle_created(
148
    event: OrderEvent,
149
) -> OrderEventData:
150
    bundle_id = event.data['ticket_bundle_id']
151
    category_id = event.data['ticket_bundle_category_id']
152
    ticket_quantity = event.data['ticket_bundle_ticket_quantity']
153
    owner_id = event.data['ticket_bundle_owner_id']
154
155
    category = ticket_category_service.find_category(category_id)
156
    category_title = category.title if (category is not None) else None
157
158
    return {
159
        'bundle_id': bundle_id,
160
        'ticket_category_title': category_title,
161
        'ticket_quantity': ticket_quantity,
162
    }
163
164
165 1
def _get_additional_data_for_ticket_bundle_revoked(
166
    event: OrderEvent, users_by_id: dict[UserID, User]
167
) -> OrderEventData:
168
    bundle_id = event.data['ticket_bundle_id']
169
170
    data = {
171
        'bundle_id': bundle_id,
172
    }
173
174
    initiator_id = event.data.get('initiator_id')
175
    if initiator_id:
176
        data['initiator'] = users_by_id[initiator_id]
177
178
    return data
179
180
181 1
def _get_additional_data_for_ticket_created(
182
    event: OrderEvent,
183
) -> OrderEventData:
184
    ticket_id = event.data['ticket_id']
185
    ticket_code = event.data['ticket_code']
186
    category_id = event.data['ticket_category_id']
187
    owner_id = event.data['ticket_owner_id']
188
189
    return {
190
        'ticket_id': ticket_id,
191
        'ticket_code': ticket_code,
192
    }
193
194
195 1
def _get_additional_data_for_ticket_revoked(
196
    event: OrderEvent, users_by_id: dict[UserID, User]
197
) -> OrderEventData:
198
    ticket_id = event.data['ticket_id']
199
    ticket_code = event.data['ticket_code']
200
201
    data = {
202
        'ticket_code': ticket_code,
203
    }
204
205
    initiator_id = event.data.get('initiator_id')
206
    if initiator_id:
207
        data['initiator'] = users_by_id[initiator_id]
208
209
    return data
210