Passed
Branch main (854eb5)
by Jochen
04:24
created

uses_any_ticket_for_party()   A

Complexity

Conditions 1

Size

Total Lines 8
Code Lines 6

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 1
CRAP Score 1.2963

Importance

Changes 0
Metric Value
cc 1
eloc 6
dl 0
loc 8
rs 10
c 0
b 0
f 0
nop 2
ccs 1
cts 3
cp 0.3333
crap 1.2963
1
"""
2
byceps.services.ticketing.ticket_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 typing import Optional, Sequence
11
12 1
from ...database import db, Pagination
13 1
from ...typing import PartyID, UserID
14
15 1
from ..party.dbmodels.party import Party as DbParty
16 1
from ..party import service as party_service
17 1
from ..seating.dbmodels.seat import Seat as DbSeat
18 1
from ..shop.order.transfer.models import OrderNumber
19 1
from ..user.dbmodels.user import User as DbUser
20
21 1
from . import event_service
22 1
from .dbmodels.category import Category as DbCategory
23 1
from .dbmodels.ticket import Ticket as DbTicket
24 1
from .dbmodels.ticket_event import TicketEvent as DbTicketEvent
25 1
from . import ticket_code_service
26 1
from .transfer.models import (
27
    TicketCategoryID,
28
    TicketCode,
29
    TicketID,
30
    TicketSaleStats,
31
)
32
33
34 1
def update_ticket_code(
35
    ticket_id: TicketID, code: str, initiator_id: UserID
36
) -> None:
37
    """Set a custom code for the ticket."""
38
    ticket = get_ticket(ticket_id)
39
40
    if not ticket_code_service.is_ticket_code_wellformed(code):
41
        raise ValueError(f'Ticket code "{code}" is not well-formed')
42
43
    old_code = ticket.code
44
45
    ticket.code = code
46
47
    event = event_service.build_event(
48
        'ticket-code-changed',
49
        ticket.id,
50
        {
51
            'old_code': old_code,
52
            'new_code': code,
53
            'initiator_id': str(initiator_id),
54
        },
55
    )
56
    db.session.add(event)
57
58
    db.session.commit()
59
60
61 1
def delete_ticket(ticket_id: TicketID) -> None:
62
    """Delete a ticket and its events."""
63 1
    db.session.query(DbTicketEvent) \
64
        .filter_by(ticket_id=ticket_id) \
65
        .delete()
66
67 1
    db.session.query(DbTicket) \
68
        .filter_by(id=ticket_id) \
69
        .delete()
70
71 1
    db.session.commit()
72
73
74 1
def find_ticket(ticket_id: TicketID) -> Optional[DbTicket]:
75
    """Return the ticket with that id, or `None` if not found."""
76 1
    return db.session.query(DbTicket).get(ticket_id)
77
78
79 1
def get_ticket(ticket_id: TicketID) -> DbTicket:
80
    """Return the ticket with that id, or raise an exception."""
81 1
    ticket = find_ticket(ticket_id)
82
83 1
    if ticket is None:
84
        raise ValueError(f'Unknown ticket ID "{ticket_id}"')
85
86 1
    return ticket
87
88
89 1
def find_ticket_by_code(
90
    party_id: PartyID, code: TicketCode
91
) -> Optional[DbTicket]:
92
    """Return the ticket with that code for that party, or `None` if not
93
    found.
94
    """
95
    return DbTicket.query \
96
        .for_party(party_id) \
97
        .filter_by(code=code) \
98
        .one_or_none()
99
100
101 1
def find_tickets(ticket_ids: set[TicketID]) -> Sequence[DbTicket]:
102
    """Return the tickets with those ids."""
103 1
    if not ticket_ids:
104
        return []
105
106 1
    return DbTicket.query \
107
        .filter(DbTicket.id.in_(ticket_ids)) \
108
        .all()
109
110
111 1
def find_tickets_created_by_order(
112
    order_number: OrderNumber,
113
) -> Sequence[DbTicket]:
114
    """Return the tickets created by this order (as it was marked as paid)."""
115 1
    return DbTicket.query \
116
        .filter_by(order_number=order_number) \
117
        .order_by(DbTicket.created_at) \
118
        .all()
119
120
121 1
def find_tickets_for_seat_manager(
122
    user_id: UserID, party_id: PartyID
123
) -> Sequence[DbTicket]:
124
    """Return the tickets for that party whose respective seats the user
125
    is entitled to manage.
126
    """
127
    return DbTicket.query \
128
        .for_party(party_id) \
129
        .filter(DbTicket.revoked == False) \
130
        .filter(
131
            (
132
                (DbTicket.seat_managed_by_id == None) &
133
                (DbTicket.owned_by_id == user_id)
134
            ) |
135
            (DbTicket.seat_managed_by_id == user_id)
136
        ) \
137
        .options(
138
            db.joinedload(DbTicket.occupied_seat),
139
        ) \
140
        .all()
141
142
143 1
def find_tickets_related_to_user(user_id: UserID) -> Sequence[DbTicket]:
144
    """Return tickets related to the user."""
145 1
    return DbTicket.query \
146
        .filter(
147
            (DbTicket.owned_by_id == user_id) |
148
            (DbTicket.seat_managed_by_id == user_id) |
149
            (DbTicket.user_managed_by_id == user_id) |
150
            (DbTicket.used_by_id == user_id)
151
        ) \
152
        .options(
153
            db.joinedload(DbTicket.occupied_seat).joinedload(DbSeat.area),
154
            db.joinedload(DbTicket.occupied_seat).joinedload(DbSeat.category),
155
            db.joinedload(DbTicket.seat_managed_by),
156
            db.joinedload(DbTicket.user_managed_by),
157
            db.joinedload(DbTicket.used_by),
158
        ) \
159
        .order_by(DbTicket.created_at) \
160
        .all()
161
162
163 1
def find_tickets_related_to_user_for_party(
164
    user_id: UserID, party_id: PartyID
165
) -> Sequence[DbTicket]:
166
    """Return tickets related to the user for the party."""
167 1
    return DbTicket.query \
168
        .for_party(party_id) \
169
        .filter(
170
            (DbTicket.owned_by_id == user_id) |
171
            (DbTicket.seat_managed_by_id == user_id) |
172
            (DbTicket.user_managed_by_id == user_id) |
173
            (DbTicket.used_by_id == user_id)
174
        ) \
175
        .options(
176
            db.joinedload(DbTicket.occupied_seat).joinedload(DbSeat.area),
177
            db.joinedload(DbTicket.occupied_seat).joinedload(DbSeat.category),
178
            db.joinedload(DbTicket.seat_managed_by),
179
            db.joinedload(DbTicket.user_managed_by),
180
            db.joinedload(DbTicket.used_by),
181
        ) \
182
        .order_by(DbTicket.created_at) \
183
        .all()
184
185
186 1
def find_tickets_used_by_user(
187
    user_id: UserID, party_id: PartyID
188
) -> Sequence[DbTicket]:
189
    """Return the tickets (if any) used by the user for that party."""
190 1
    return DbTicket.query \
191
        .for_party(party_id) \
192
        .filter(DbTicket.used_by_id == user_id) \
193
        .filter(DbTicket.revoked == False) \
194
        .outerjoin(DbSeat) \
195
        .options(
196
            db.joinedload(DbTicket.occupied_seat).joinedload(DbSeat.area),
197
        ) \
198
        .order_by(DbSeat.coord_x, DbSeat.coord_y) \
199
        .all()
200
201
202 1
def uses_any_ticket_for_party(user_id: UserID, party_id: PartyID) -> bool:
203
    """Return `True` if the user uses any ticket for that party."""
204
    q = DbTicket.query \
205
        .for_party(party_id) \
206
        .filter(DbTicket.used_by_id == user_id) \
207
        .filter(DbTicket.revoked == False)
208
209
    return db.session.query(q.exists()).scalar()
210
211
212 1
def get_ticket_users_for_party(party_id: PartyID) -> set[UserID]:
213
    """Return the IDs of the users of tickets for that party."""
214
    rows = db.session \
215
        .query(DbTicket.used_by_id) \
216
        .filter(DbTicket.party_id == party_id) \
217
        .filter(DbTicket.revoked == False) \
218
        .filter(DbTicket.used_by_id != None) \
219
        .all()
220
221
    return {row[0] for row in rows}
222
223
224 1
def select_ticket_users_for_party(
225
    user_ids: set[UserID], party_id: PartyID
226
) -> set[UserID]:
227
    """Return the IDs of those users that use a ticket for that party."""
228 1
    if not user_ids:
229
        return set()
230
231 1
    q = DbTicket.query \
232
        .for_party(party_id) \
233
        .filter(DbTicket.used_by_id == DbUser.id) \
234
        .filter(DbTicket.revoked == False)
235
236 1
    rows = db.session.query(DbUser.id) \
237
        .filter(q.exists()) \
238
        .filter(DbUser.id.in_(user_ids)) \
239
        .all()
240
241 1
    return {row[0] for row in rows}
242
243
244 1
def get_ticket_with_details(ticket_id: TicketID) -> Optional[DbTicket]:
245
    """Return the ticket with that id, or `None` if not found."""
246 1
    return db.session.query(DbTicket) \
247
        .options(
248
            db.joinedload(DbTicket.category),
249
            db.joinedload(DbTicket.occupied_seat).joinedload(DbSeat.area),
250
            db.joinedload(DbTicket.owned_by),
251
            db.joinedload(DbTicket.seat_managed_by),
252
            db.joinedload(DbTicket.user_managed_by),
253
        ) \
254
        .get(ticket_id)
255
256
257 1
def get_tickets_with_details_for_party_paginated(
258
    party_id: PartyID,
259
    page: int,
260
    per_page: int,
261
    *,
262
    search_term: Optional[str] = None,
263
    only_category_id: Optional[TicketCategoryID] = None,
264
) -> Pagination:
265
    """Return the party's tickets to show on the specified page."""
266 1
    query = DbTicket.query \
267
        .for_party(party_id) \
268
        .join(DbCategory) \
269
        .options(
270
            db.joinedload(DbTicket.category),
271
            db.joinedload(DbTicket.owned_by),
272
            db.joinedload(DbTicket.occupied_seat).joinedload(DbSeat.area),
273
        )
274
275 1
    if search_term:
276
        ilike_pattern = f'%{search_term}%'
277
        query = query \
278
            .filter(DbTicket.code.ilike(ilike_pattern))
279
280 1
    if only_category_id:
281
        query = query \
282
            .filter(DbCategory.id == str(only_category_id))
283
284 1
    return query \
285
        .order_by(DbTicket.created_at) \
286
        .paginate(page, per_page)
287
288
289 1
def get_ticket_count_by_party_id() -> dict[PartyID, int]:
290
    """Return ticket count (including 0) per party, indexed by party ID."""
291
    party = db.aliased(DbParty)
292
293
    subquery = db.session \
294
        .query(
295
            db.func.count(DbTicket.id)
296
        ) \
297
        .join(DbCategory) \
298
        .filter(DbCategory.party_id == party.id) \
299
        .filter(DbTicket.revoked == False) \
300
        .subquery() \
301
        .as_scalar()
302
303
    party_ids_and_ticket_counts = db.session \
304
        .query(
305
            party.id,
306
            subquery
307
        ) \
308
        .all()
309
310
    return dict(party_ids_and_ticket_counts)
311
312
313 1
def count_revoked_tickets_for_party(party_id: PartyID) -> int:
314
    """Return the number of revoked tickets for that party."""
315 1
    return DbTicket.query \
316
        .for_party(party_id) \
317
        .filter(DbTicket.revoked == True) \
318
        .count()
319
320
321 1
def count_sold_tickets_for_party(party_id: PartyID) -> int:
322
    """Return the number of "sold" (i.e. generated and not revoked)
323
    tickets for that party.
324
    """
325 1
    return DbTicket.query \
326
        .for_party(party_id) \
327
        .filter(DbTicket.revoked == False) \
328
        .count()
329
330
331 1
def count_tickets_checked_in_for_party(party_id: PartyID) -> int:
332
    """Return the number tickets for that party that were used to check
333
    in their respective user.
334
    """
335 1
    return DbTicket.query \
336
        .for_party(party_id) \
337
        .filter(DbTicket.user_checked_in == True) \
338
        .count()
339
340
341 1
def get_ticket_sale_stats(party_id: PartyID) -> TicketSaleStats:
342
    """Return the number of maximum and sold tickets, respectively."""
343 1
    party = party_service.get_party(party_id)
344
345 1
    sold = count_sold_tickets_for_party(party.id)
346
347 1
    return TicketSaleStats(
348
        tickets_max=party.max_ticket_quantity,
349
        tickets_sold=sold,
350
    )
351