byceps.services.seating.seat_group_service.find_seat_group_occupied_by_ticket_bundle() - Code Metrics - Inspection of "Release potentially occupied seat group on ticket..." - byceps/byceps - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( 0acbe1...472285 )

by Jochen

created 2022-03-07 04:22 UTC

find_seat_group_occupied_by_ticket_bundle() A

↳ Parent: byceps.services.seating.seat_group_service

Complexity

Conditions

Size

Total Lines	10
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	6
nop	1
dl	0
loc	10
ccs	2
cts	2
cp	1
crap	1
rs	10
c	0
b	0
f	0

"""
byceps.services.seating.seat_group_service
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

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

from typing import Optional, Sequence

from sqlalchemy import select

from ...database import db
from ...typing import PartyID

from ..ticketing.dbmodels.ticket import Ticket as DbTicket
from ..ticketing.dbmodels.ticket_bundle import TicketBundle as DbTicketBundle
from ..ticketing.transfer.models import TicketBundleID, TicketCategoryID

from .dbmodels.seat import Seat as DbSeat
from .dbmodels.seat_group import (
    Occupancy as DbSeatGroupOccupancy,
    SeatGroup as DbSeatGroup,
    SeatGroupAssignment as DbSeatGroupAssignment,
)
from .transfer.models import SeatID, SeatGroupID


def create_seat_group(
    party_id: PartyID,
    ticket_category_id: TicketCategoryID,
    title: str,
    seats: Sequence[DbSeat],
    *,
    commit: bool = True,
) -> DbSeatGroup:
    """Create a seat group and assign the given seats."""
    seat_quantity = len(seats)
    if seat_quantity == 0:
        raise ValueError("No seats specified.")

    ticket_category_ids = {seat.category_id for seat in seats}
    if len(ticket_category_ids) != 1 or (
        ticket_category_id not in ticket_category_ids
    ):
        raise ValueError("Seats' ticket category IDs do not match the group's.")

    group = DbSeatGroup(party_id, ticket_category_id, seat_quantity, title)
    db.session.add(group)

    for seat in seats:
        assignment = DbSeatGroupAssignment(group, seat)
        db.session.add(assignment)

    if commit:
        db.session.commit()

    return group


def occupy_seat_group(
    seat_group: DbSeatGroup, ticket_bundle: DbTicketBundle
) -> DbSeatGroupOccupancy:
    """Occupy the seat group with that ticket bundle."""
    seats = seat_group.seats
    tickets = ticket_bundle.tickets

    _ensure_group_is_available(seat_group)
    _ensure_categories_match(seat_group, ticket_bundle)
    _ensure_quantities_match(seat_group, ticket_bundle)
    _ensure_actual_quantities_match(seats, tickets)

    occupancy = DbSeatGroupOccupancy(seat_group.id, ticket_bundle.id)
    db.session.add(occupancy)

    _occupy_seats(seats, tickets)

    db.session.commit()

    return occupancy


def switch_seat_group(
    occupancy: DbSeatGroupOccupancy, to_group: DbSeatGroup
) -> None:
    """Switch ticket bundle to another seat group."""
    ticket_bundle = occupancy.ticket_bundle
    tickets = ticket_bundle.tickets
    seats = to_group.seats

    _ensure_group_is_available(to_group)
    _ensure_categories_match(to_group, ticket_bundle)
    _ensure_quantities_match(to_group, ticket_bundle)
    _ensure_actual_quantities_match(seats, tickets)

    occupancy.seat_group_id = to_group.id

    _occupy_seats(seats, tickets)

    db.session.commit()


def _ensure_group_is_available(seat_group: DbSeatGroup) -> None:
    """Raise an error if the seat group is occupied."""
    occupancy = find_occupancy_for_seat_group(seat_group.id)
    if occupancy is not None:
        raise ValueError('Seat group is already occupied.')


def _ensure_categories_match(
    seat_group: DbSeatGroup, ticket_bundle: DbTicketBundle
) -> None:
    """Raise an error if the seat group's and the ticket bundle's
    categories don't match.
    """
    if seat_group.ticket_category_id != ticket_bundle.ticket_category_id:
        raise ValueError('Seat and ticket categories do not match.')


def _ensure_quantities_match(
    seat_group: DbSeatGroup, ticket_bundle: DbTicketBundle
) -> None:
    """Raise an error if the seat group's and the ticket bundle's
    quantities don't match.
    """
    if seat_group.seat_quantity != ticket_bundle.ticket_quantity:
        raise ValueError('Seat and ticket quantities do not match.')


def _ensure_actual_quantities_match(
    seats: Sequence[DbSeat], tickets: Sequence[DbTicket]
) -> None:
    """Raise an error if the totals of seats and tickets don't match."""
    if len(seats) != len(tickets):
        raise ValueError(
            'The actual quantities of seats and tickets ' 'do not match.'
        )


def _occupy_seats(seats: Sequence[DbSeat], tickets: Sequence[DbTicket]) -> None:
    """Occupy all seats in the group with all tickets from the bundle."""
    seats = _sort_seats(seats)
    tickets = _sort_tickets(tickets)

    for seat, ticket in zip(seats, tickets):
        ticket.occupied_seat = seat


def _sort_seats(seats: Sequence[DbSeat]) -> Sequence[DbSeat]:
    """Create a list of the seats sorted by their respective coordinates."""
    return list(sorted(seats, key=lambda s: (s.coord_x, s.coord_y)))


def _sort_tickets(tickets: Sequence[DbTicket]) -> Sequence[DbTicket]:
    """Create a list of the tickets sorted by creation time (ascending)."""
    return list(sorted(tickets, key=lambda t: t.created_at))


def release_seat_group(seat_group_id: SeatGroupID) -> None:
    """Release a seat group so it becomes available again."""
    occupancy = find_occupancy_for_seat_group(seat_group_id)
    if occupancy is None:
        raise ValueError('Seat group is not occupied.')

    for ticket in occupancy.ticket_bundle.tickets:
        ticket.occupied_seat = None

    db.session.delete(occupancy)

    db.session.commit()


def count_seat_groups_for_party(party_id: PartyID) -> int:
    """Return the number of seat groups for that party."""
    return db.session \
        .query(DbSeatGroup) \
        .filter_by(party_id=party_id) \
        .count()


def find_seat_group(seat_group_id: SeatGroupID) -> Optional[DbSeatGroup]:
    """Return the seat group with that id, or `None` if not found."""
    return db.session.get(DbSeatGroup, seat_group_id)


def find_seat_group_occupied_by_ticket_bundle(
    ticket_bundle_id: TicketBundleID,
) -> Optional[SeatGroupID]:
    """Return the ID of the seat group occupied by that ticket bundle,
    or `None` if not found.
    """
    return db.session.execute(
        select(DbSeatGroupOccupancy.seat_group_id)
        .filter_by(ticket_bundle_id=ticket_bundle_id)
    ).one_or_none()


def find_occupancy_for_seat_group(
    seat_group_id: SeatGroupID,
) -> Optional[DbSeatGroupOccupancy]:
    """Return the occupancy for that seat group, or `None` if not found."""
    return db.session.execute(
        select(DbSeatGroupOccupancy)
        .filter_by(seat_group_id=seat_group_id)
    ).one_or_none()


def get_all_seat_groups_for_party(party_id: PartyID) -> Sequence[DbSeatGroup]:
    """Return all seat groups for that party."""
    return db.session \
        .query(DbSeatGroup) \
        .filter_by(party_id=party_id) \
        .all()


def is_seat_part_of_a_group(seat_id: SeatID) -> bool:
    """Return whether or not the seat is part of a seat group."""
    return db.session.execute(
        select(
            select(DbSeatGroupAssignment).filter_by(seat_id=seat_id).exists()
        )
    ).scalar_one()


1		"""
2		byceps.services.seating.seat_group_service
3		~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
4
5		:Copyright: 2014-2022 Jochen Kupperschmidt
6		:License: Revised BSD (see `LICENSE` file for details)
7		"""
8
9	1	from typing import Optional, Sequence
10
11	1	from sqlalchemy import select
12
13	1	from ...database import db
14	1	from ...typing import PartyID
15
16	1	from ..ticketing.dbmodels.ticket import Ticket as DbTicket
17	1	from ..ticketing.dbmodels.ticket_bundle import TicketBundle as DbTicketBundle
18	1	from ..ticketing.transfer.models import TicketBundleID, TicketCategoryID
19
20	1	from .dbmodels.seat import Seat as DbSeat
21	1	from .dbmodels.seat_group import (
22		Occupancy as DbSeatGroupOccupancy,
23		SeatGroup as DbSeatGroup,
24		SeatGroupAssignment as DbSeatGroupAssignment,
25		)
26	1	from .transfer.models import SeatID, SeatGroupID
27
28
29	1	def create_seat_group(
30		party_id: PartyID,
31		ticket_category_id: TicketCategoryID,
32		title: str,
33		seats: Sequence[DbSeat],
34		*,
35		commit: bool = True,
36		) -> DbSeatGroup:
37		"""Create a seat group and assign the given seats."""
38		seat_quantity = len(seats)
39		if seat_quantity == 0:
40		raise ValueError("No seats specified.")
41
42		ticket_category_ids = {seat.category_id for seat in seats}
43		if len(ticket_category_ids) != 1 or (
44		ticket_category_id not in ticket_category_ids
45		):
46		raise ValueError("Seats' ticket category IDs do not match the group's.")
47
48		group = DbSeatGroup(party_id, ticket_category_id, seat_quantity, title)
49		db.session.add(group)
50
51		for seat in seats:
52		assignment = DbSeatGroupAssignment(group, seat)
53		db.session.add(assignment)
54
55		if commit:
56		db.session.commit()
57
58		return group
59
60
61	1	def occupy_seat_group(
62		seat_group: DbSeatGroup, ticket_bundle: DbTicketBundle
63		) -> DbSeatGroupOccupancy:
64		"""Occupy the seat group with that ticket bundle."""
65		seats = seat_group.seats
66		tickets = ticket_bundle.tickets
67
68		_ensure_group_is_available(seat_group)
69		_ensure_categories_match(seat_group, ticket_bundle)
70		_ensure_quantities_match(seat_group, ticket_bundle)
71		_ensure_actual_quantities_match(seats, tickets)
72
73		occupancy = DbSeatGroupOccupancy(seat_group.id, ticket_bundle.id)
74		db.session.add(occupancy)
75
76		_occupy_seats(seats, tickets)
77
78		db.session.commit()
79
80		return occupancy
81
82
83	1	def switch_seat_group(
84		occupancy: DbSeatGroupOccupancy, to_group: DbSeatGroup
85		) -> None:
86		"""Switch ticket bundle to another seat group."""
87		ticket_bundle = occupancy.ticket_bundle
88		tickets = ticket_bundle.tickets
89		seats = to_group.seats
90
91		_ensure_group_is_available(to_group)
92		_ensure_categories_match(to_group, ticket_bundle)
93		_ensure_quantities_match(to_group, ticket_bundle)
94		_ensure_actual_quantities_match(seats, tickets)
95
96		occupancy.seat_group_id = to_group.id
97
98		_occupy_seats(seats, tickets)
99
100		db.session.commit()
101
102
103	1	def _ensure_group_is_available(seat_group: DbSeatGroup) -> None:
104		"""Raise an error if the seat group is occupied."""
105		occupancy = find_occupancy_for_seat_group(seat_group.id)
106		if occupancy is not None:
107		raise ValueError('Seat group is already occupied.')
108
109
110	1	def _ensure_categories_match(
111		seat_group: DbSeatGroup, ticket_bundle: DbTicketBundle
112		) -> None:
113		"""Raise an error if the seat group's and the ticket bundle's
114		categories don't match.
115		"""
116		if seat_group.ticket_category_id != ticket_bundle.ticket_category_id:
117		raise ValueError('Seat and ticket categories do not match.')
118
119
120	1	def _ensure_quantities_match(
121		seat_group: DbSeatGroup, ticket_bundle: DbTicketBundle
122		) -> None:
123		"""Raise an error if the seat group's and the ticket bundle's
124		quantities don't match.
125		"""
126		if seat_group.seat_quantity != ticket_bundle.ticket_quantity:
127		raise ValueError('Seat and ticket quantities do not match.')
128
129
130	1	def _ensure_actual_quantities_match(
131		seats: Sequence[DbSeat], tickets: Sequence[DbTicket]
132		) -> None:
133		"""Raise an error if the totals of seats and tickets don't match."""
134		if len(seats) != len(tickets):
135		raise ValueError(
136		'The actual quantities of seats and tickets ' 'do not match.'
137		)
138
139
140	1	def _occupy_seats(seats: Sequence[DbSeat], tickets: Sequence[DbTicket]) -> None:
141		"""Occupy all seats in the group with all tickets from the bundle."""
142		seats = _sort_seats(seats)
143		tickets = _sort_tickets(tickets)
144
145		for seat, ticket in zip(seats, tickets):
146		ticket.occupied_seat = seat
147
148
149	1	def _sort_seats(seats: Sequence[DbSeat]) -> Sequence[DbSeat]:
150		"""Create a list of the seats sorted by their respective coordinates."""
151		return list(sorted(seats, key=lambda s: (s.coord_x, s.coord_y)))
152
153
154	1	def _sort_tickets(tickets: Sequence[DbTicket]) -> Sequence[DbTicket]:
155		"""Create a list of the tickets sorted by creation time (ascending)."""
156		return list(sorted(tickets, key=lambda t: t.created_at))
157
158
159	1	def release_seat_group(seat_group_id: SeatGroupID) -> None:
160		"""Release a seat group so it becomes available again."""
161		occupancy = find_occupancy_for_seat_group(seat_group_id)
162		if occupancy is None:
163		raise ValueError('Seat group is not occupied.')
164
165		for ticket in occupancy.ticket_bundle.tickets:
166		ticket.occupied_seat = None
167
168		db.session.delete(occupancy)
169
170		db.session.commit()
171
172
173	1	def count_seat_groups_for_party(party_id: PartyID) -> int:
174		"""Return the number of seat groups for that party."""
175	1	return db.session \
176		.query(DbSeatGroup) \
177		.filter_by(party_id=party_id) \
178		.count()
179
180
181	1	def find_seat_group(seat_group_id: SeatGroupID) -> Optional[DbSeatGroup]:
182		"""Return the seat group with that id, or `None` if not found."""
183		return db.session.get(DbSeatGroup, seat_group_id)
184
185
186	1	def find_seat_group_occupied_by_ticket_bundle(
187		ticket_bundle_id: TicketBundleID,
188		) -> Optional[SeatGroupID]:
189		"""Return the ID of the seat group occupied by that ticket bundle,
190		or `None` if not found.
191		"""
192	1	return db.session.execute(
193		select(DbSeatGroupOccupancy.seat_group_id)
194		.filter_by(ticket_bundle_id=ticket_bundle_id)
195		).one_or_none()
196
197
198	1	def find_occupancy_for_seat_group(
199		seat_group_id: SeatGroupID,
200		) -> Optional[DbSeatGroupOccupancy]:
201		"""Return the occupancy for that seat group, or `None` if not found."""
202		return db.session.execute(
203		select(DbSeatGroupOccupancy)
204		.filter_by(seat_group_id=seat_group_id)
205		).one_or_none()
206
207
208	1	def get_all_seat_groups_for_party(party_id: PartyID) -> Sequence[DbSeatGroup]:
209		"""Return all seat groups for that party."""
210	1	return db.session \
211		.query(DbSeatGroup) \
212		.filter_by(party_id=party_id) \
213		.all()
214
215
216	1	def is_seat_part_of_a_group(seat_id: SeatID) -> bool:
217		"""Return whether or not the seat is part of a seat group."""
218	1	return db.session.execute(
219		select(
220		select(DbSeatGroupAssignment).filter_by(seat_id=seat_id).exists()
221		)
222		).scalar_one()
223

byceps / byceps

Push — main ( 0acbe1...472285 )

find_seat_group_occupied_by_ticket_bundle() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like