Passed
Push — main ( 42cd7b...1909da )
by Jochen
01:59
created

_get_db_group()   A

Complexity

Conditions 1

Size

Total Lines 3
Code Lines 2

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 0
CRAP Score 2

Importance

Changes 0
Metric Value
cc 1
eloc 2
nop 1
dl 0
loc 3
ccs 0
cts 0
cp 0
crap 2
rs 10
c 0
b 0
f 0
1
"""
2
byceps.services.user_group.user_group_service
3
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
4
5
:Copyright: 2014-2024 Jochen Kupperschmidt
6
:License: Revised BSD (see `LICENSE` file for details)
7
"""
8
9 1
from datetime import datetime
10
from uuid import UUID
11 1
12
from sqlalchemy import select
13 1
14
from byceps.database import db
15 1
from byceps.services.party.models import Party, PartyID
16 1
from byceps.services.user import user_service
17 1
from byceps.services.user.models.user import User, UserID
18
19 1
from . import user_group_domain_service
20
from .dbmodels import DbUserGroup
21
from .models import UserGroup
22 1
23
24
def create_group(
25
    party: Party,
26
    creator: User,
27
    title: str,
28
    description: str | None,
29
) -> UserGroup:
30
    """Create a group."""
31
    created_at = datetime.utcnow()
32
33
    group = user_group_domain_service.create_group(
34
        party, created_at, creator, title, description=description
35
    )
36
37 1
    db_group = DbUserGroup(
38
        group.id,
39
        group.party_id,
40
        group.created_at,
41
        group.creator.id,
42
        group.title,
43
        group.description,
44
    )
45
46
    db.session.add(db_group)
47
    db.session.commit()
48
49
    return group
50
51
52
def update_group(group: UserGroup, title: str, description: str) -> None:
53
    """Update a group."""
54
    db_group = _get_db_group(group.id)
55
    if db_group is None:
56
        raise ValueError(f'Unknown user group ID "{group.id}"')
57
58
    db_group.title = title
59
    db_group.description = description or None
60
61
    db.session.commit()
62
63
64
def is_title_available(party_id: PartyID, title: str) -> bool:
65
    """Check if the title is yet unused."""
66
    return not db.session.scalar(
67
        select(
68
            db.exists()
69
            .where(DbUserGroup.party_id == party_id)
70
            .where(db.func.lower(DbUserGroup.title) == title.lower())
71
        )
72
    )
73
74
75
def find_group(group_id: UUID) -> UserGroup | None:
76
    """Return the group, if found."""
77
    db_group = _get_db_group(group_id)
78
79
    if db_group is None:
80
        return None
81
82
    creator = user_service.get_user(db_group.creator_id, include_avatar=True)
83
    return _db_entity_to_group(db_group, {creator.id: creator})
84
85
86
def _get_db_group(group_id: UUID) -> DbUserGroup | None:
87
    """Return the group, if found."""
88
    return db.session.get(DbUserGroup, group_id)
89
90
91
def get_groups_for_party(party_id: PartyID) -> list[UserGroup]:
92
    """Return user groups for a party."""
93
    db_groups = db.session.scalars(select(DbUserGroup)).all()
94
95
    user_ids = {db_group.creator_id for db_group in db_groups}
96
    users_by_id = user_service.get_users_indexed_by_id(
97
        user_ids, include_avatars=True
98
    )
99
100
    return [
101
        _db_entity_to_group(db_group, users_by_id) for db_group in db_groups
102
    ]
103
104
105
def _db_entity_to_group(
106
    db_group: DbUserGroup, users_by_id: dict[UserID, User]
107
) -> UserGroup:
108
    creator = users_by_id[db_group.creator_id]
109
110
    return UserGroup(
111
        id=db_group.id,
112
        party_id=db_group.party_id,
113
        created_at=db_group.created_at,
114
        creator=creator,
115
        title=db_group.title,
116
        description=db_group.description,
117
    )
118