~homeworkprod/byceps

ref: 4237b3ec9496efe95dcce82bea3207ab9de4d520 byceps/byceps/services/ticketing/ticket_revocation_service.py -rw-r--r-- 1.8 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
"""
byceps.services.ticketing.ticket_revocation_service
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

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

from typing import Optional, Set

from ...database import db
from ...typing import UserID

from . import event_service, ticket_seat_management_service
from .event_service import TicketEvent
from . import ticket_service
from .transfer.models import TicketID


def revoke_ticket(
    ticket_id: TicketID, initiator_id: UserID, *, reason: Optional[str] = None
) -> None:
    """Revoke the ticket."""
    ticket = ticket_service.find_ticket(ticket_id)

    if ticket is None:
        raise ValueError('Unknown ticket ID.')

    # Release seat.
    if ticket.occupied_seat_id:
        ticket_seat_management_service.release_seat(ticket.id, initiator_id)

    ticket.revoked = True

    event = _build_ticket_revoked_event(ticket.id, initiator_id, reason)
    db.session.add(event)

    db.session.commit()


def revoke_tickets(
    ticket_ids: Set[TicketID],
    initiator_id: UserID,
    *,
    reason: Optional[str] = None,
) -> None:
    """Revoke the tickets."""
    tickets = ticket_service.find_tickets(ticket_ids)

    # Release seats.
    for ticket in tickets:
        if ticket.occupied_seat_id:
            ticket_seat_management_service.release_seat(ticket.id, initiator_id)

    for ticket in tickets:
        ticket.revoked = True

        event = _build_ticket_revoked_event(ticket.id, initiator_id, reason)
        db.session.add(event)

    db.session.commit()


def _build_ticket_revoked_event(
    ticket_id: TicketID, initiator_id: UserID, reason: Optional[str] = None
) -> TicketEvent:
    data = {
        'initiator_id': str(initiator_id),
    }

    if reason:
        data['reason'] = reason

    return event_service.build_event('ticket-revoked', ticket_id, data)