_is_current_user_seating_admin()   A
last analyzed

Complexity

Conditions 1

Size

Total Lines 2
Code Lines 2

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 1
CRAP Score 1.125

Importance

Changes 0
Metric Value
cc 1
eloc 2
nop 0
dl 0
loc 2
ccs 1
cts 2
cp 0.5
crap 1.125
rs 10
c 0
b 0
f 0
1
"""
2
byceps.blueprints.site.seating.views
3
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
4
5
:Copyright: 2014-2024 Jochen Kupperschmidt
6
:License: Revised BSD (see `LICENSE` file for details)
7
"""
8
9 1
from typing import Any
10
11 1
from flask import abort, g, request
12 1
from flask_babel import gettext
13
14 1
from byceps.blueprints.site.site.navigation import subnavigation_for_view
15 1
from byceps.services.seating import (
16
    seat_service,
17
    seating_area_service,
18
    seating_area_tickets_service,
19
)
20 1
from byceps.services.seating.models import Seat, SeatID, SeatingArea
21 1
from byceps.services.ticketing import (
22
    errors as ticketing_errors,
23
    ticket_seat_management_service,
24
    ticket_service,
25
)
26 1
from byceps.services.ticketing.dbmodels.ticket import DbTicket
27 1
from byceps.services.ticketing.models.ticket import TicketID
28 1
from byceps.util.authz import has_current_user_permission
29 1
from byceps.util.framework.blueprint import create_blueprint
30 1
from byceps.util.framework.flash import flash_error, flash_success
31 1
from byceps.util.framework.templating import templated
32 1
from byceps.util.views import login_required, redirect_to, respond_no_content
33
34
35 1
blueprint = create_blueprint('seating', __name__)
36
37
38 1
@blueprint.get('/')
39 1
@templated
40 1
@subnavigation_for_view('seating_plan')
41 1
def index():
42
    """List areas."""
43
    if g.party is None:
44
        # No party is configured for the current site.
45
        abort(404)
46
47
    areas = seating_area_service.get_areas_for_party(g.party_id)
48
    if len(areas) == 1:
49
        return _render_view_area(areas[0])
50
51
    areas_with_utilization = (
52
        seating_area_service.get_areas_with_seat_utilization(g.party_id)
53
    )
54
    if not areas_with_utilization:
55
        abort(404)
56
57
    seat_utilizations = [awu[1] for awu in areas_with_utilization]
58
    total_seat_utilization = seat_service.aggregate_seat_utilizations(
59
        seat_utilizations
60
    )
61
62
    return {
63
        'areas_with_utilization': areas_with_utilization,
64
        'total_seat_utilization': total_seat_utilization,
65
    }
66
67
68 1
@blueprint.get('/areas/<slug>')
69 1
def view_area(slug):
70
    """View area."""
71
    if g.party is None:
72
        # No party is configured for the current site.
73
        abort(404)
74
75
    area = seating_area_service.find_area_for_party_by_slug(g.party_id, slug)
76
    if area is None:
77
        abort(404)
78
79
    return _render_view_area(area)
80
81
82 1
@templated('site/seating/view_area')
83 1
@subnavigation_for_view('seating_plan')
84 1
def _render_view_area(area: SeatingArea) -> dict[str, Any]:
85
    seat_management_enabled = _is_seat_management_enabled()
86
87
    seats_with_tickets = seat_service.get_seats_with_tickets_for_area(area.id)
88
89
    users_by_id = seating_area_tickets_service.get_users(seats_with_tickets, [])
90
91
    seats_and_tickets = seating_area_tickets_service.get_seats_and_tickets(
92
        seats_with_tickets, users_by_id
93
    )
94
95
    seat_utilization = seat_service.get_seat_utilization(g.party_id)
96
97
    return {
98
        'area': area,
99
        'seat_management_enabled': seat_management_enabled,
100
        'seats_and_tickets': seats_and_tickets,
101
        'seat_utilization': seat_utilization,
102
        'manage_mode': False,
103
    }
104
105
106 1
@blueprint.get('/areas/<slug>/manage_seats')
107 1
@login_required
108 1
@templated('site/seating/view_area')
109 1
@subnavigation_for_view('seating_plan')
110 1
def manage_seats_in_area(slug):
111
    """Manage seats for assigned tickets in area."""
112
    if not _is_seat_management_enabled():
113
        flash_error(
114
            gettext('Seat reservations cannot be changed at this time.')
115
        )
116
        return redirect_to('.view_area', slug=slug)
117
118
    area = seating_area_service.find_area_for_party_by_slug(g.party_id, slug)
119
    if area is None:
120
        abort(404)
121
122
    seat_management_enabled = _is_seat_management_enabled()
123
124
    seat_manager_id = None
125
    selected_ticket_id = None
126
    selected_ticket = None
127
128
    if _is_current_user_seating_admin():
129
        selected_ticket = _get_selected_ticket()
130
        if selected_ticket is not None:
131
            seat_manager_id = selected_ticket.get_seat_manager().id
132
            selected_ticket_id = selected_ticket.id
133
        elif seat_management_enabled:
134
            seat_manager_id = g.user.id
135
136
    elif seat_management_enabled:
137
        seat_manager_id = g.user.id
138
139
    seats_with_tickets = seat_service.get_seats_with_tickets_for_area(area.id)
140
141
    if seat_manager_id is not None:
142
        tickets = ticket_service.get_tickets_for_seat_manager(
143
            seat_manager_id, g.party_id
144
        )
145
    else:
146
        tickets = []
147
148
    users_by_id = seating_area_tickets_service.get_users(
149
        seats_with_tickets, tickets
150
    )
151
152
    seats_and_tickets = seating_area_tickets_service.get_seats_and_tickets(
153
        seats_with_tickets, users_by_id
154
    )
155
156
    if seat_management_enabled:
157
        managed_tickets = list(
158
            seating_area_tickets_service.get_managed_tickets(
159
                tickets, users_by_id
160
            )
161
        )
162
    else:
163
        managed_tickets = []
164
165
    seat_utilization = seat_service.get_seat_utilization(g.party_id)
166
167
    return {
168
        'area': area,
169
        'seats_and_tickets': seats_and_tickets,
170
        'seat_utilization': seat_utilization,
171
        'manage_mode': True,
172
        'seat_management_enabled': seat_management_enabled,
173
        'managed_tickets': managed_tickets,
174
        'selected_ticket_id': selected_ticket_id,
175
    }
176
177
178 1
def _get_selected_ticket():
179
    selected_ticket = None
180
181
    selected_ticket_id_arg = request.args.get('ticket_id')
182
    if selected_ticket_id_arg:
183
        selected_ticket = ticket_service.find_ticket(selected_ticket_id_arg)
184
        if selected_ticket is None:
185
            flash_error(
186
                gettext(
187
                    'Ticket ID "%(selected_ticket_id_arg)s" not found.',
188
                    selected_ticket_id_arg=selected_ticket_id_arg,
189
                )
190
            )
191
192
    if (selected_ticket is not None) and selected_ticket.revoked:
193
        flash_error(
194
            gettext(
195
                'Ticket "%(selected_ticket_code)s" is revoked.',
196
                selected_ticket_code=selected_ticket.code,
197
            )
198
        )
199
        selected_ticket = None
200
201
    return selected_ticket
202
203
204 1
@blueprint.post('/ticket/<uuid:ticket_id>/seat/<uuid:seat_id>')
205 1
@login_required
206 1
@respond_no_content
207 1
def occupy_seat(ticket_id, seat_id):
208
    """Use ticket to occupy seat."""
209
    if not _is_seat_management_enabled():
210
        flash_error(
211
            gettext('Seat reservations cannot be changed at this time.')
212
        )
213
        return
214
215
    ticket = _get_ticket_or_404(ticket_id)
216
217
    manager = g.user
218
219
    if (
220
        not ticket.is_seat_managed_by(manager.id)
221
        and not _is_current_user_seating_admin()
222
    ):
223
        flash_error(
224
            gettext(
225
                'You are not authorized to manage the seat for ticket %(ticket_code)s.',
226
                ticket_code=ticket.code,
227
            )
228
        )
229
        return
230
231
    seat = _get_seat_or_404(seat_id)
232
233
    if ticket_service.find_ticket_occupying_seat(seat.id) is not None:
234
        flash_error(
235
            gettext(
236
                '%(seat_label)s is already occupied.', seat_label=seat.label
237
            )
238
        )
239
        return
240
241
    try:
242
        occupy_seat_result = ticket_seat_management_service.occupy_seat(
243
            ticket.id, seat.id, manager.id
244
        )
245
    except ValueError:
246
        abort(404)
247
248
    if occupy_seat_result.is_err():
249
        err = occupy_seat_result.unwrap_err()
250
        if isinstance(
251
            err, ticketing_errors.SeatChangeDeniedForBundledTicketError
252
        ):
253
            flash_error(
254
                gettext(
255
                    'Ticket %(ticket_code)s belongs to a bundle and cannot be managed separately.',
256
                    ticket_code=ticket.code,
257
                )
258
            )
259
        elif isinstance(err, ticketing_errors.TicketCategoryMismatchError):
260
            flash_error(
261
                gettext(
262
                    'Ticket %(ticket_code)s and seat "%(seat_label)s" belong to different categories.',
263
                    ticket_code=ticket.code,
264
                    seat_label=seat.label,
265
                )
266
            )
267
        else:
268
            flash_error(gettext('An unexpected error occurred.'))
269
270
        return
271
272
    flash_success(
273
        gettext(
274
            '%(seat_label)s has been occupied with ticket %(ticket_code)s.',
275
            seat_label=seat.label,
276
            ticket_code=ticket.code,
277
        )
278
    )
279
280
281 1
@blueprint.delete('/ticket/<uuid:ticket_id>/seat')
282 1
@login_required
283 1
@respond_no_content
284 1
def release_seat(ticket_id):
285
    """Release the seat."""
286
    if not _is_seat_management_enabled():
287
        flash_error(
288
            gettext('Seat reservations cannot be changed at this time.')
289
        )
290
        return
291
292
    ticket = _get_ticket_or_404(ticket_id)
293
294
    if not ticket.occupied_seat:
295
        flash_error(
296
            gettext(
297
                'Ticket %(ticket_code)s occupies no seat.',
298
                ticket_code=ticket.code,
299
            )
300
        )
301
        return
302
303
    manager = g.user
304
305
    if (
306
        not ticket.is_seat_managed_by(manager.id)
307
        and not _is_current_user_seating_admin()
308
    ):
309
        flash_error(
310
            gettext(
311
                'You are not authorized to manage the seat for ticket %(ticket_code)s.',
312
                ticket_code=ticket.code,
313
            )
314
        )
315
        return
316
317
    seat = ticket.occupied_seat
318
319
    release_seat_result = ticket_seat_management_service.release_seat(
320
        ticket.id, manager.id
321
    )
322
323
    if release_seat_result.is_err():
324
        err = release_seat_result.unwrap_err()
325
        if isinstance(
326
            err, ticketing_errors.SeatChangeDeniedForBundledTicketError
327
        ):
328
            flash_error(
329
                gettext(
330
                    'Ticket %(ticket_code)s belongs to a bundle and cannot be managed separately.',
331
                    ticket_code=ticket.code,
332
                )
333
            )
334
        else:
335
            flash_error(gettext('An unexpected error occurred.'))
336
337
        return
338
339
    flash_success(
340
        gettext('%(seat_label)s has been released.', seat_label=seat.label)
341
    )
342
343
344 1
def _is_seat_management_enabled():
345
    if not g.user.authenticated:
346
        return False
347
348
    if g.party_id is None:
349
        return False
350
351
    if _is_current_user_seating_admin():
352
        return True
353
354
    return g.party.seat_management_enabled
355
356
357 1
def _is_current_user_seating_admin() -> bool:
358
    return has_current_user_permission('ticketing.administrate_seat_occupancy')
359
360
361 1
def _get_ticket_or_404(ticket_id: TicketID) -> DbTicket:
362
    ticket = ticket_service.find_ticket(ticket_id)
363
364
    if (ticket is None) or ticket.revoked:
365
        abort(404)
366
367
    return ticket
368
369
370 1
def _get_seat_or_404(seat_id: SeatID) -> Seat:
371
    seat = seat_service.find_seat(seat_id)
372
373
    if seat is None:
374
        abort(404)
375
376
    return seat
377