byceps.services.ticketing.ticket_service.uses_any_ticket_for_party() - Code Metrics - Inspection of "854eb576321a419e4bcbff29042e010f176ff84d" - byceps/byceps - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Branch — main (854eb5)

by Jochen

created 2021-07-19 11:44 UTC

uses_any_ticket_for_party() A

↳ Parent: byceps.services.ticketing.ticket_service

Complexity

Conditions

Size

Total Lines	8
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	1.2963

Importance

Changes

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

"""
byceps.services.ticketing.ticket_service
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

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

from __future__ import annotations
from typing import Optional, Sequence

from ...database import db, Pagination
from ...typing import PartyID, UserID

from ..party.dbmodels.party import Party as DbParty
from ..party import service as party_service
from ..seating.dbmodels.seat import Seat as DbSeat
from ..shop.order.transfer.models import OrderNumber
from ..user.dbmodels.user import User as DbUser

from . import event_service
from .dbmodels.category import Category as DbCategory
from .dbmodels.ticket import Ticket as DbTicket
from .dbmodels.ticket_event import TicketEvent as DbTicketEvent
from . import ticket_code_service
from .transfer.models import (
    TicketCategoryID,
    TicketCode,
    TicketID,
    TicketSaleStats,
)


def update_ticket_code(
    ticket_id: TicketID, code: str, initiator_id: UserID
) -> None:
    """Set a custom code for the ticket."""
    ticket = get_ticket(ticket_id)

    if not ticket_code_service.is_ticket_code_wellformed(code):
        raise ValueError(f'Ticket code "{code}" is not well-formed')

    old_code = ticket.code

    ticket.code = code

    event = event_service.build_event(
        'ticket-code-changed',
        ticket.id,
        {
            'old_code': old_code,
            'new_code': code,
            'initiator_id': str(initiator_id),
        },
    )
    db.session.add(event)

    db.session.commit()


def delete_ticket(ticket_id: TicketID) -> None:
    """Delete a ticket and its events."""
    db.session.query(DbTicketEvent) \
        .filter_by(ticket_id=ticket_id) \
        .delete()

    db.session.query(DbTicket) \
        .filter_by(id=ticket_id) \
        .delete()

    db.session.commit()


def find_ticket(ticket_id: TicketID) -> Optional[DbTicket]:
    """Return the ticket with that id, or `None` if not found."""
    return db.session.query(DbTicket).get(ticket_id)


def get_ticket(ticket_id: TicketID) -> DbTicket:
    """Return the ticket with that id, or raise an exception."""
    ticket = find_ticket(ticket_id)

    if ticket is None:
        raise ValueError(f'Unknown ticket ID "{ticket_id}"')

    return ticket


def find_ticket_by_code(
    party_id: PartyID, code: TicketCode
) -> Optional[DbTicket]:
    """Return the ticket with that code for that party, or `None` if not
    found.
    """
    return DbTicket.query \
        .for_party(party_id) \
        .filter_by(code=code) \
        .one_or_none()


def find_tickets(ticket_ids: set[TicketID]) -> Sequence[DbTicket]:
    """Return the tickets with those ids."""
    if not ticket_ids:
        return []

    return DbTicket.query \
        .filter(DbTicket.id.in_(ticket_ids)) \
        .all()


def find_tickets_created_by_order(
    order_number: OrderNumber,
) -> Sequence[DbTicket]:
    """Return the tickets created by this order (as it was marked as paid)."""
    return DbTicket.query \
        .filter_by(order_number=order_number) \
        .order_by(DbTicket.created_at) \
        .all()


def find_tickets_for_seat_manager(
    user_id: UserID, party_id: PartyID
) -> Sequence[DbTicket]:
    """Return the tickets for that party whose respective seats the user
    is entitled to manage.
    """
    return DbTicket.query \
        .for_party(party_id) \
        .filter(DbTicket.revoked == False) \
        .filter(
            (
                (DbTicket.seat_managed_by_id == None) &
                (DbTicket.owned_by_id == user_id)
            ) |
            (DbTicket.seat_managed_by_id == user_id)
        ) \
        .options(
            db.joinedload(DbTicket.occupied_seat),
        ) \
        .all()


def find_tickets_related_to_user(user_id: UserID) -> Sequence[DbTicket]:
    """Return tickets related to the user."""
    return DbTicket.query \
        .filter(
            (DbTicket.owned_by_id == user_id) |
            (DbTicket.seat_managed_by_id == user_id) |
            (DbTicket.user_managed_by_id == user_id) |
            (DbTicket.used_by_id == user_id)
        ) \
        .options(
            db.joinedload(DbTicket.occupied_seat).joinedload(DbSeat.area),
            db.joinedload(DbTicket.occupied_seat).joinedload(DbSeat.category),
            db.joinedload(DbTicket.seat_managed_by),
            db.joinedload(DbTicket.user_managed_by),
            db.joinedload(DbTicket.used_by),
        ) \
        .order_by(DbTicket.created_at) \
        .all()


def find_tickets_related_to_user_for_party(
    user_id: UserID, party_id: PartyID
) -> Sequence[DbTicket]:
    """Return tickets related to the user for the party."""
    return DbTicket.query \
        .for_party(party_id) \
        .filter(
            (DbTicket.owned_by_id == user_id) |
            (DbTicket.seat_managed_by_id == user_id) |
            (DbTicket.user_managed_by_id == user_id) |
            (DbTicket.used_by_id == user_id)
        ) \
        .options(
            db.joinedload(DbTicket.occupied_seat).joinedload(DbSeat.area),
            db.joinedload(DbTicket.occupied_seat).joinedload(DbSeat.category),
            db.joinedload(DbTicket.seat_managed_by),
            db.joinedload(DbTicket.user_managed_by),
            db.joinedload(DbTicket.used_by),
        ) \
        .order_by(DbTicket.created_at) \
        .all()


def find_tickets_used_by_user(
    user_id: UserID, party_id: PartyID
) -> Sequence[DbTicket]:
    """Return the tickets (if any) used by the user for that party."""
    return DbTicket.query \
        .for_party(party_id) \
        .filter(DbTicket.used_by_id == user_id) \
        .filter(DbTicket.revoked == False) \
        .outerjoin(DbSeat) \
        .options(
            db.joinedload(DbTicket.occupied_seat).joinedload(DbSeat.area),
        ) \
        .order_by(DbSeat.coord_x, DbSeat.coord_y) \
        .all()


def uses_any_ticket_for_party(user_id: UserID, party_id: PartyID) -> bool:
    """Return `True` if the user uses any ticket for that party."""
    q = DbTicket.query \
        .for_party(party_id) \
        .filter(DbTicket.used_by_id == user_id) \
        .filter(DbTicket.revoked == False)

    return db.session.query(q.exists()).scalar()


def get_ticket_users_for_party(party_id: PartyID) -> set[UserID]:
    """Return the IDs of the users of tickets for that party."""
    rows = db.session \
        .query(DbTicket.used_by_id) \
        .filter(DbTicket.party_id == party_id) \
        .filter(DbTicket.revoked == False) \
        .filter(DbTicket.used_by_id != None) \
        .all()

    return {row[0] for row in rows}


def select_ticket_users_for_party(
    user_ids: set[UserID], party_id: PartyID
) -> set[UserID]:
    """Return the IDs of those users that use a ticket for that party."""
    if not user_ids:
        return set()

    q = DbTicket.query \
        .for_party(party_id) \
        .filter(DbTicket.used_by_id == DbUser.id) \
        .filter(DbTicket.revoked == False)

    rows = db.session.query(DbUser.id) \
        .filter(q.exists()) \
        .filter(DbUser.id.in_(user_ids)) \
        .all()

    return {row[0] for row in rows}


def get_ticket_with_details(ticket_id: TicketID) -> Optional[DbTicket]:
    """Return the ticket with that id, or `None` if not found."""
    return db.session.query(DbTicket) \
        .options(
            db.joinedload(DbTicket.category),
            db.joinedload(DbTicket.occupied_seat).joinedload(DbSeat.area),
            db.joinedload(DbTicket.owned_by),
            db.joinedload(DbTicket.seat_managed_by),
            db.joinedload(DbTicket.user_managed_by),
        ) \
        .get(ticket_id)


def get_tickets_with_details_for_party_paginated(
    party_id: PartyID,
    page: int,
    per_page: int,
    *,
    search_term: Optional[str] = None,
    only_category_id: Optional[TicketCategoryID] = None,
) -> Pagination:
    """Return the party's tickets to show on the specified page."""
    query = DbTicket.query \
        .for_party(party_id) \
        .join(DbCategory) \
        .options(
            db.joinedload(DbTicket.category),
            db.joinedload(DbTicket.owned_by),
            db.joinedload(DbTicket.occupied_seat).joinedload(DbSeat.area),
        )

    if search_term:
        ilike_pattern = f'%{search_term}%'
        query = query \
            .filter(DbTicket.code.ilike(ilike_pattern))

    if only_category_id:
        query = query \
            .filter(DbCategory.id == str(only_category_id))

    return query \
        .order_by(DbTicket.created_at) \
        .paginate(page, per_page)


def get_ticket_count_by_party_id() -> dict[PartyID, int]:
    """Return ticket count (including 0) per party, indexed by party ID."""
    party = db.aliased(DbParty)

    subquery = db.session \
        .query(
            db.func.count(DbTicket.id)
        ) \
        .join(DbCategory) \
        .filter(DbCategory.party_id == party.id) \
        .filter(DbTicket.revoked == False) \
        .subquery() \
        .as_scalar()

    party_ids_and_ticket_counts = db.session \
        .query(
            party.id,
            subquery
        ) \
        .all()

    return dict(party_ids_and_ticket_counts)


def count_revoked_tickets_for_party(party_id: PartyID) -> int:
    """Return the number of revoked tickets for that party."""
    return DbTicket.query \
        .for_party(party_id) \
        .filter(DbTicket.revoked == True) \
        .count()


def count_sold_tickets_for_party(party_id: PartyID) -> int:
    """Return the number of "sold" (i.e. generated and not revoked)
    tickets for that party.
    """
    return DbTicket.query \
        .for_party(party_id) \
        .filter(DbTicket.revoked == False) \
        .count()


def count_tickets_checked_in_for_party(party_id: PartyID) -> int:
    """Return the number tickets for that party that were used to check
    in their respective user.
    """
    return DbTicket.query \
        .for_party(party_id) \
        .filter(DbTicket.user_checked_in == True) \
        .count()


def get_ticket_sale_stats(party_id: PartyID) -> TicketSaleStats:
    """Return the number of maximum and sold tickets, respectively."""
    party = party_service.get_party(party_id)

    sold = count_sold_tickets_for_party(party.id)

    return TicketSaleStats(
        tickets_max=party.max_ticket_quantity,
        tickets_sold=sold,
    )


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

byceps / byceps

Branch — main (854eb5)

uses_any_ticket_for_party() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like