Passed
Push — master ( 1a2c01...af3a31 )
by Jochen
02:18
created

get_ticket_sale_stats()   A

Complexity

Conditions 1

Size

Total Lines 9
Code Lines 6

Duplication

Lines 0
Ratio 0 %

Importance

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