Completed
Push — main ( c065c4...1fd2bb )
by Jochen
04:54
created

byceps.blueprints.seating.service   A

Complexity

Total Complexity 16

Size/Duplication

Total Lines 122
Duplicated Lines 0 %

Test Coverage

Coverage 51.67%

Importance

Changes 0
Metric Value
eloc 79
dl 0
loc 122
ccs 31
cts 60
cp 0.5167
rs 10
c 0
b 0
f 0
wmc 16

7 Functions

Rating   Name   Duplication   Size   Complexity  
A _get_ticket_user_ids() 0 5 3
A get_seats() 0 22 3
A get_managed_tickets() 0 13 3
A get_users() 0 7 1
A _get_ticket_users_by_id() 0 4 1
A _get_seat_tickets() 0 4 3
A _find_ticket_user() 0 7 2
1
"""
2
byceps.blueprints.seating.service
3
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
4
5
:Copyright: 2006-2020 Jochen Kupperschmidt
6
:License: Modified BSD, see LICENSE for details.
7
"""
8
9 1
from dataclasses import dataclass
10 1
from itertools import chain
11 1
from typing import Dict, Iterator, Optional, Sequence
12
13 1
from ...services.seating.models.seat import Seat
14 1
from ...services.seating.transfer.models import SeatID
15 1
from ...services.ticketing.models.ticket import Ticket as DbTicket
16 1
from ...services.ticketing.transfer.models import (
17
    TicketCategoryID,
18
    TicketCode,
19
    TicketID,
20
)
21 1
from ...services.user import service as user_service
22 1
from ...services.user.transfer.models import User
23 1
from ...typing import UserID
24
25
26 1
@dataclass(frozen=True)
27
class ManagedTicket:
28 1
    id: TicketID
29 1
    code: TicketCode
30 1
    category_label: str
31 1
    user: User
32 1
    occupied_seat_label: str
33
34
35 1
@dataclass(frozen=True)
36
class Seat:
37 1
    id: SeatID
38 1
    coord_x: int
39 1
    coord_y: int
40 1
    category_id: TicketCategoryID
41 1
    label: str
42 1
    type_: str
43 1
    ticket: ManagedTicket
44
45
46 1
def get_users(
47
    seats: Sequence[Seat], managed_tickets: Sequence[DbTicket]
48
) -> Dict[UserID, User]:
49
    seat_tickets = _get_seat_tickets(seats)
50
    tickets = chain(seat_tickets, managed_tickets)
51
52
    return _get_ticket_users_by_id(tickets)
53
54
55 1
def _get_seat_tickets(seats: Sequence[Seat]) -> Iterator[DbTicket]:
56
    for seat in seats:
57
        if seat.has_user:
58
            yield seat.occupied_by_ticket
59
60
61 1
def _get_ticket_users_by_id(tickets: Sequence[DbTicket]) -> Dict[UserID, User]:
62
    user_ids = set(_get_ticket_user_ids(tickets))
63
    users = user_service.find_users(user_ids, include_avatars=True)
64
    return user_service.index_users_by_id(users)
65
66
67 1
def _get_ticket_user_ids(tickets: Sequence[DbTicket]) -> Iterator[UserID]:
68
    for ticket in tickets:
69
        user_id = ticket.used_by_id
70
        if user_id is not None:
71
            yield user_id
72
73
74 1
def get_seats(
75
    seats: Sequence[Seat], users_by_id: Dict[UserID, User]
76
) -> Iterator[Seat]:
77
    for seat in seats:
78
        if seat.is_occupied:
79
            ticket = seat.occupied_by_ticket
80
            user = _find_ticket_user(ticket, users_by_id)
81
82
            managed_ticket = ManagedTicket(
83
                ticket.id, ticket.code, ticket.category.title, user, None
84
            )
85
        else:
86
            managed_ticket = None
87
88
        yield Seat(
89
            seat.id,
90
            seat.coord_x,
91
            seat.coord_y,
92
            seat.category_id,
93
            seat.label,
94
            seat.type_,
95
            managed_ticket,
96
        )
97
98
99 1
def get_managed_tickets(
100
    managed_tickets: Sequence[DbTicket], users_by_id: Dict[UserID, User]
101
) -> Iterator[ManagedTicket]:
102
    for ticket in managed_tickets:
103
        user = _find_ticket_user(ticket, users_by_id)
104
105
        if ticket.occupied_seat_id is not None:
106
            seat_label = ticket.occupied_seat.label
107
        else:
108
            seat_label = None
109
110
        yield ManagedTicket(
111
            ticket.id, ticket.code, ticket.category.title, user, seat_label
112
        )
113
114
115 1
def _find_ticket_user(
116
    ticket: DbTicket, users_by_id: Dict[UserID, User]
117
) -> Optional[User]:
118
    if ticket.used_by_id is None:
119
        return None
120
121
    return users_by_id[ticket.used_by_id]
122