byceps.services.ticketing.ticket_service.get_ticket_sale_stats() - Code Metrics - Inspection of "Freshen up presentation of admin shops list" - byceps/byceps - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 1a2c01...af3a31 )

by Jochen

created 2019-11-18 01:27 UTC

byceps.services.ticketing.ticket_service.get_ticket_sale_stats() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	9
Code Lines	6

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	6
nop	1
dl	0
loc	9
rs	10
c	0
b	0
f	0

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

:Copyright: 2006-2019 Jochen Kupperschmidt
:License: Modified BSD, see LICENSE for details.
"""

from typing import Dict, Optional, Sequence, Set

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

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

from .models.category import Category as DbCategory
from .models.ticket import Ticket as DbTicket
from .transfer.models import TicketCode, TicketID, TicketSaleStats


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


def find_ticket_by_code(code: TicketCode) -> Optional[DbTicket]:
    """Return the ticket with that code, or `None` if not found."""
    return DbTicket.query \
        .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('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('occupied_seat').joinedload('area'),
            db.joinedload('occupied_seat').joinedload('category'),
            db.joinedload('seat_managed_by'),
            db.joinedload('user_managed_by'),
            db.joinedload('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('occupied_seat').joinedload('area'),
            db.joinedload('occupied_seat').joinedload('category'),
            db.joinedload('seat_managed_by'),
            db.joinedload('user_managed_by'),
            db.joinedload('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('occupied_seat').joinedload('area'),
        ) \
        .order_by(DbSeat.coord_x, DbSeat.coord_y) \
        .all()


def find_tickets_used_by_user_simplified(
    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) \
        .options(
            db.joinedload('occupied_seat').joinedload('area'),
        ) \
        .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 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 DbTicket.query \
        .options(
            db.joinedload('category'),
            db.joinedload('occupied_seat').joinedload('area'),
            db.joinedload('owned_by'),
            db.joinedload('seat_managed_by'),
            db.joinedload('user_managed_by'),
        ) \
        .get(ticket_id)


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

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

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


def get_tickets_in_use_for_party_paginated(
    party_id: PartyID,
    page: int,
    per_page: int,
    *,
    search_term: Optional[str] = None,
) -> Pagination:
    """Return the party's tickets which have a user assigned."""
    ticket_user = db.aliased(DbUser)

    query = DbTicket.query \
        .for_party(party_id) \
        .filter(DbTicket.revoked == False) \
        .filter(DbTicket.used_by_id.isnot(None))

    if search_term:
        query = query \
            .filter(ticket_user.screen_name.ilike(f'%{search_term}%'))

    return query \
        .join(ticket_user, DbTicket.used_by_id == ticket_user.id) \
        .order_by(db.func.lower(ticket_user.screen_name), 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_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_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-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

byceps / byceps

Push — master ( 1a2c01...af3a31 )

byceps.services.ticketing.ticket_service.get_ticket_sale_stats() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like