~homeworkprod/byceps

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

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

from typing import Dict, Optional, Sequence

from ...database import db
from ...typing import PartyID

from .models.category import Category as DbCategory
from .models.ticket import Ticket as DbTicket
from .transfer.models import TicketCategory, TicketCategoryID


def create_category(party_id: PartyID, title: str) -> TicketCategory:
    """Create a category."""
    category = DbCategory(party_id, title)

    db.session.add(category)
    db.session.commit()

    return _db_entity_to_category(category)


def delete_category(category_id: TicketCategoryID) -> None:
    """Delete a category."""
    db.session.query(DbCategory) \
        .filter_by(id=category_id) \
        .delete()
    db.session.commit()


def count_categories_for_party(party_id: PartyID) -> int:
    """Return the number of categories for that party."""
    return DbCategory.query \
        .for_party(party_id) \
        .count()


def find_category(category_id: TicketCategoryID) -> Optional[TicketCategory]:
    """Return the category with that ID, or `None` if not found."""
    category = DbCategory.query.get(category_id)

    return _db_entity_to_category(category)


def get_categories_for_party(party_id: PartyID) -> Sequence[TicketCategory]:
    """Return all categories for that party."""
    categories = DbCategory.query \
        .for_party(party_id) \
        .all()

    return [_db_entity_to_category(category) for category in categories]


def get_categories_with_ticket_counts_for_party(
    party_id: PartyID,
) -> Dict[TicketCategory, int]:
    """Return all categories with ticket counts for that party."""
    category = db.aliased(DbCategory)

    subquery = db.session \
        .query(
            db.func.count(DbTicket.id)
        ) \
        .join(DbCategory) \
        .filter(DbCategory.id == category.id) \
        .filter(DbTicket.revoked == False) \
        .subquery() \
        .as_scalar()

    rows = db.session \
        .query(
            category,
            subquery
        ) \
        .filter(category.party_id == party_id) \
        .group_by(category.id) \
        .all()

    return {
        _db_entity_to_category(category): ticket_count
        for category, ticket_count in rows
    }


def _db_entity_to_category(category: DbCategory) -> TicketCategory:
    return TicketCategory(
        category.id,
        category.party_id,
        category.title,
    )