~homeworkprod/byceps

ref: 4237b3ec9496efe95dcce82bea3207ab9de4d520 byceps/byceps/services/attendance/service.py -rw-r--r-- 3.6 KiB
4237b3ec — Jochen Kupperschmidt Move ticketing blueprint into `site` subpackage 1 year, 11 months ago
                                                                                
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
"""
byceps.services.attendance.service
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

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

from collections import defaultdict
from typing import Dict, Iterable, List, Optional, Set, Tuple

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

from ..ticketing.models.ticket import Category as DbCategory, Ticket as DbTicket
from ..user.models.user import User as DbUser

from .transfer.models import Attendee, AttendeeTicket


def get_attendees_paginated(
    party_id: PartyID,
    page: int,
    per_page: int,
    *,
    search_term: Optional[str] = None,
) -> Pagination:
    """Return the party's ticket users with tickets and seats."""
    users_paginated = _get_users_paginated(
        party_id, page, per_page, search_term=search_term
    )
    users = users_paginated.items
    user_ids = {u.id for u in users}

    tickets = _get_tickets_for_users(party_id, user_ids)
    tickets_by_user_id = _index_tickets_by_user_id(tickets)

    attendees = list(_generate_attendees(users, tickets_by_user_id))

    users_paginated.items = attendees
    return users_paginated


def _get_users_paginated(
    party_id: PartyID,
    page: int,
    per_page: int,
    *,
    search_term: Optional[str] = None,
) -> Pagination:
    # Drop revoked tickets here already to avoid users without tickets
    # being included in the list.
    query = DbUser.query \
        .distinct() \
        .options(
            db.load_only('id', 'screen_name', 'deleted'),
            db.joinedload('avatar_selection').joinedload('avatar'),
        ) \
        .join(DbTicket, DbTicket.used_by_id == DbUser.id) \
        .filter(DbTicket.revoked == False) \
        .join(DbCategory).filter(DbCategory.party_id == party_id)

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

    query = query \
        .order_by(db.func.lower(DbUser.screen_name))

    return paginate(query, page, per_page)


def _get_tickets_for_users(
    party_id: PartyID, user_ids: Set[UserID]
) -> List[DbTicket]:
    return DbTicket.query \
        .options(
            db.joinedload('category'),
            db.joinedload('occupied_seat').joinedload('area'),
        ) \
        .for_party(party_id) \
        .filter(DbTicket.used_by_id.in_(user_ids)) \
        .filter(DbTicket.revoked == False) \
        .all()


def _index_tickets_by_user_id(
    tickets: Iterable[DbTicket],
) -> Dict[UserID, Set[DbTicket]]:
    tickets_by_user_id = defaultdict(set)
    for ticket in tickets:
        tickets_by_user_id[ticket.used_by_id].add(ticket)
    return tickets_by_user_id


def _generate_attendees(
    users: Iterable[DbUser], tickets_by_user_id: Dict[UserID, Set[DbTicket]]
) -> Iterable[Attendee]:
    for user in users:
        tickets = tickets_by_user_id[user.id]
        attendee_tickets = _to_attendee_tickets(tickets)
        yield Attendee(user, attendee_tickets)


def _to_attendee_tickets(tickets: Iterable[DbTicket]) -> List[AttendeeTicket]:
    attendee_tickets = [
        AttendeeTicket(t.occupied_seat, t.user_checked_in) for t in tickets
    ]
    attendee_tickets.sort(key=_get_attendee_ticket_sort_key)
    return attendee_tickets


def _get_attendee_ticket_sort_key(
    attendee_ticket: AttendeeTicket,
) -> Tuple[bool, str, bool]:
    return (
        # List tickets with occupied seat first.
        attendee_ticket.seat is None,

        # Sort by seat label.
        attendee_ticket.seat.label if attendee_ticket.seat else None,

        # List checked in tickets first.
        not attendee_ticket.checked_in,
    )