~homeworkprod/byceps

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

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

from typing import Optional, Sequence

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

from ..shop.order.transfer.models import OrderNumber

from .models.ticket import Ticket as DbTicket
from .models.ticket_bundle import TicketBundle as DbTicketBundle
from .ticket_creation_service import build_tickets
from .ticket_revocation_service import (
    _build_ticket_revoked_event as build_ticket_revoked_event,
)
from .transfer.models import TicketBundleID, TicketCategoryID


def create_bundle(
    category_id: TicketCategoryID,
    ticket_quantity: int,
    owned_by_id: UserID,
    *,
    order_number: Optional[OrderNumber] = None,
) -> DbTicketBundle:
    """Create a ticket bundle and the given quantity of tickets."""
    if ticket_quantity < 1:
        raise ValueError('Ticket quantity must be positive.')

    bundle = DbTicketBundle(category_id, ticket_quantity, owned_by_id)
    db.session.add(bundle)

    tickets = list(
        build_tickets(
            category_id,
            owned_by_id,
            ticket_quantity,
            bundle=bundle,
            order_number=order_number,
        )
    )
    db.session.add_all(tickets)

    db.session.commit()

    return bundle


def revoke_bundle(
    bundle_id: TicketBundleID,
    initiator_id: UserID,
    *,
    reason: Optional[str] = None,
) -> None:
    """Revoke the tickets included in this bundle."""
    bundle = find_bundle(bundle_id)

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

    for ticket in bundle.tickets:
        ticket.revoked = True

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

    db.session.commit()


def delete_bundle(bundle_id: TicketBundleID) -> None:
    """Delete a bundle and the tickets assigned to it."""
    bundle = find_bundle(bundle_id)

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

    db.session.query(DbTicket) \
        .filter_by(bundle_id=bundle_id) \
        .delete()

    db.session.query(DbTicketBundle) \
        .filter_by(id=bundle_id) \
        .delete()

    db.session.commit()


def find_bundle(bundle_id: TicketBundleID) -> Optional[DbTicketBundle]:
    """Return the ticket bundle with that id, or `None` if not found."""
    return DbTicketBundle.query.get(bundle_id)


def find_tickets_for_bundle(bundle_id: TicketBundleID) -> Sequence[DbTicket]:
    """Return all tickets included in this bundle."""
    return DbTicket.query \
        .filter(DbTicket.bundle_id == bundle_id) \
        .all()