~homeworkprod/byceps

ref: 948380c701bcc5014efe74b559fe3e744e73de92 byceps/byceps/services/tourney/tourney_service.py -rw-r--r-- 4.5 KiB
948380c7 — Jochen Kupperschmidt Introduce `_get_order_entity` 6 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
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
"""
byceps.services.tourney.tourney_service
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

:Copyright: 2006-2018 Jochen Kupperschmidt
:License: Revised BSD (see `LICENSE` file for details)
"""

from __future__ import annotations
from datetime import datetime
from typing import Optional

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

from ..party import service as party_service

from . import category_service
from .dbmodels.participant import Participant as DbParticipant
from .dbmodels.tourney import Tourney as DbTourney
from .dbmodels.tourney_category import TourneyCategory as DbTourneyCategory
from .transfer.models import (
    Tourney,
    TourneyCategoryID,
    TourneyID,
    TourneyWithCategory,
)


def create_tourney(
    party_id: PartyID,
    title: str,
    category_id: TourneyCategoryID,
    max_participant_count: int,
    starts_at: datetime,
    *,
    subtitle: Optional[str] = None,
    logo_url: Optional[str] = None,
) -> Tourney:
    """Create a tourney."""
    party = party_service.get_party(party_id)

    category = category_service.find_category(category_id)
    if category is None:
        raise ValueError(f'Unknown tourney category ID "{category_id}"')

    tourney = DbTourney(
        party.id,
        title,
        category.id,
        max_participant_count,
        starts_at,
        subtitle=subtitle,
        logo_url=logo_url,
    )

    db.session.add(tourney)
    db.session.commit()

    return _db_entity_to_tourney(tourney)


def update_tourney(
    tourney_id: TourneyID,
    title: str,
    subtitle: Optional[str],
    logo_url: Optional[str],
    category_id: TourneyCategoryID,
    max_participant_count: int,
    starts_at: datetime,
) -> Tourney:
    """Update tourney."""
    tourney = _get_db_tourney(tourney_id)

    tourney.title = title
    tourney.subtitle = subtitle
    tourney.logo_url = logo_url
    tourney.category_id = category_id
    tourney.max_participant_count = max_participant_count
    tourney.starts_at = starts_at

    db.session.commit()

    return _db_entity_to_tourney(tourney)


def delete_tourney(tourney_id: TourneyID) -> None:
    """Delete a tourney."""
    tourney = get_tourney(tourney_id)

    db.session.query(DbTourney) \
        .filter_by(id=tourney_id) \
        .delete()

    db.session.commit()


def find_tourney(tourney_id: int) -> Optional[Tourney]:
    """Return the tourney with that id, or `None` if not found."""
    tourney = _find_db_tourney(tourney_id)

    if tourney is None:
        return None

    return _db_entity_to_tourney(tourney)


def _find_db_tourney(tourney_id: TourneyID) -> Optional[DbTourney]:
    return DbTourney.query.get(tourney_id)


def get_tourney(tourney_id: int) -> Tourney:
    """Return the tourney with that ID.

    Raise an exception if not found.
    """
    tourney = find_tourney(tourney_id)

    if tourney is None:
        raise ValueError(f'Unknown tourney ID "{tourney_id}"')

    return tourney


def _get_db_tourney(tourney_id: int) -> DbTourney:
    tourney = _find_db_tourney(tourney_id)

    if tourney is None:
        raise ValueError(f'Unknown tourney ID "{tourney_id}"')

    return tourney


def get_tourneys_for_party(party_id: PartyID) -> list[TourneyWithCategory]:
    """Return the tourneys for that party."""
    rows = db.session \
        .query(DbTourney, DbTourneyCategory, db.func.count(DbParticipant.id)) \
        .join(DbTourneyCategory) \
        .join(DbParticipant, isouter=True) \
        .filter(DbTourney.party_id == party_id) \
        .group_by(DbTourney.id, DbTourneyCategory) \
        .all()

    return [_to_tourney_with_category(row[0], row[1], row[2]) for row in rows]


def _db_entity_to_tourney(
    tourney: DbTourney,
    current_participant_count: int = -1,
) -> Tourney:
    return Tourney(
        id=tourney.id,
        party_id=tourney.party_id,
        title=tourney.title,
        subtitle=tourney.subtitle,
        logo_url=tourney.logo_url,
        category_id=tourney.category_id,
        current_participant_count=current_participant_count,
        max_participant_count=tourney.max_participant_count,
        starts_at=tourney.starts_at,
    )


def _to_tourney_with_category(
    db_tourney: DbTourney,
    db_category: DbTourneyCategory,
    current_participant_count: int = -1,
) -> Tourney:
    tourney = _db_entity_to_tourney(db_tourney, current_participant_count)
    category = category_service._db_entity_to_category(db_category)

    return TourneyWithCategory.from_tourney_and_category(
        tourney, category, current_participant_count
    )