Completed
Push — main ( cd2364...dc9c2e )
by Jochen
03:37
created

byceps.services.site.service.get_sites_for_brand()   A

Complexity

Conditions 1

Size

Total Lines 7
Code Lines 5

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 3
CRAP Score 1

Importance

Changes 0
Metric Value
cc 1
eloc 5
nop 1
dl 0
loc 7
ccs 3
cts 3
cp 1
crap 1
rs 10
c 0
b 0
f 0
1
"""
2
byceps.services.site.service
3
~~~~~~~~~~~~~~~~~~~~~~~~~~~~
4
5
:Copyright: 2006-2020 Jochen Kupperschmidt
6
:License: Modified BSD, see LICENSE for details.
7
"""
8
9 1
import dataclasses
10 1
from typing import List, Optional, Union
11
12 1
from ...database import db
13 1
from ...typing import BrandID, PartyID
14
15 1
from ..board.transfer.models import BoardID
16 1
from ..brand import service as brand_service
17 1
from ..news.transfer.models import ChannelID as NewsChannelID
18 1
from ..shop.storefront.transfer.models import StorefrontID
19
20 1
from .models.site import Site as DbSite
21 1
from .models.setting import Setting as DbSetting
22 1
from .transfer.models import Site, SiteID, SiteWithBrand
23
24
25 1
class UnknownSiteId(Exception):
26 1
    pass
27
28
29 1
def create_site(
30
    site_id: SiteID,
31
    title: str,
32
    server_name: str,
33
    brand_id: BrandID,
34
    email_config_id: str,
35
    *,
36
    enabled: bool = False,
37
    user_account_creation_enabled: bool = False,
38
    login_enabled: bool = False,
39
    party_id: Optional[PartyID] = None,
40
    news_channel_id: Optional[NewsChannelID] = None,
41
    board_id: Optional[BoardID] = None,
42
    storefront_id: Optional[StorefrontID] = None,
43
) -> Site:
44
    """Create a site for that party."""
45 1
    site = DbSite(
46
        site_id,
47
        title,
48
        server_name,
49
        brand_id,
50
        email_config_id,
51
        enabled,
52
        user_account_creation_enabled,
53
        login_enabled,
54
        party_id=party_id,
55
        news_channel_id=news_channel_id,
56
        board_id=board_id,
57
        storefront_id=storefront_id,
58
    )
59
60 1
    db.session.add(site)
61 1
    db.session.commit()
62
63 1
    return _db_entity_to_site(site)
64
65
66 1
def update_site(
67
    site_id: SiteID,
68
    title: str,
69
    server_name: str,
70
    brand_id: BrandID,
71
    email_config_id: str,
72
    party_id: Optional[PartyID],
73
    enabled: bool,
74
    user_account_creation_enabled: bool,
75
    login_enabled: bool,
76
    news_channel_id: Optional[NewsChannelID],
77
    board_id: Optional[BoardID],
78
    storefront_id: Optional[StorefrontID],
79
    archived: bool,
80
) -> Site:
81
    """Update the site."""
82
    site = DbSite.query.get(site_id)
83
84
    if site is None:
85
        raise UnknownSiteId(site_id)
86
87
    site.title = title
88
    site.server_name = server_name
89
    site.brand_id = brand_id
90
    site.email_config_id = email_config_id
91
    site.party_id = party_id
92
    site.enabled = enabled
93
    site.user_account_creation_enabled = user_account_creation_enabled
94
    site.login_enabled = login_enabled
95
    site.news_channel_id = news_channel_id
96
    site.board_id = board_id
97
    site.storefront_id = storefront_id
98
    site.archived = archived
99
100
    db.session.commit()
101
102
    return _db_entity_to_site(site)
103
104
105 1
def delete_site(site_id: SiteID) -> None:
106
    """Delete a site."""
107 1
    db.session.query(DbSetting) \
108
        .filter_by(site_id=site_id) \
109
        .delete()
110
111 1
    db.session.query(DbSite) \
112
        .filter_by(id=site_id) \
113
        .delete()
114
115 1
    db.session.commit()
116
117
118 1
def find_site(site_id: SiteID) -> Optional[Site]:
119
    """Return the site with that ID, or `None` if not found."""
120 1
    site = DbSite.query.get(site_id)
121
122 1
    if site is None:
123 1
        return None
124
125 1
    return _db_entity_to_site(site)
126
127
128 1
def get_site(site_id: SiteID) -> Site:
129
    """Return the site with that ID."""
130 1
    site = find_site(site_id)
131
132 1
    if site is None:
133
        raise UnknownSiteId(site_id)
134
135 1
    return site
136
137
138 1
def get_all_sites() -> List[Site]:
139
    """Return all sites."""
140 1
    sites = DbSite.query.all()
141
142 1
    return [_db_entity_to_site(site) for site in sites]
143
144
145 1
def get_sites_for_brand(brand_id: BrandID) -> List[Site]:
146
    """Return the sites for that brand."""
147 1
    sites = DbSite.query \
148
        .filter_by(brand_id=brand_id) \
149
        .all()
150
151 1
    return [_db_entity_to_site(site) for site in sites]
152
153
154 1
def get_current_sites(*, include_brands: bool = True) -> List[Union[Site, SiteWithBrand]]:
155
    """Return all "current" (i.e. enabled and not archived) sites."""
156
    query = DbSite.query
157
158
    if include_brands:
159
        query = query.options(db.joinedload('brand'))
160
161
    sites = query \
162
        .filter_by(enabled=True) \
163
        .filter_by(archived=False) \
164
        .all()
165
166
    if include_brands:
167
        transform = _db_entity_to_site_with_brand
168
    else:
169
        transform = _db_entity_to_site
170
171
    return [transform(site) for site in sites]
172
173
174 1
def _db_entity_to_site(site: DbSite) -> Site:
175 1
    return Site(
176
        site.id,
177
        site.title,
178
        site.server_name,
179
        site.brand_id,
180
        site.email_config_id,
181
        site.party_id,
182
        site.enabled,
183
        site.user_account_creation_enabled,
184
        site.login_enabled,
185
        site.news_channel_id,
186
        site.board_id,
187
        site.storefront_id,
188
        site.archived,
189
    )
190
191
192 1
def _db_entity_to_site_with_brand(site_entity: DbSite) -> SiteWithBrand:
193
    site = _db_entity_to_site(site_entity)
194
    brand = brand_service._db_entity_to_brand(site_entity.brand)
195
196
    return SiteWithBrand(*dataclasses.astuple(site), brand=brand)
197