Passed
Push — main ( 7f92a7...2ef5e3 )
by Jochen
07:43
created

_get_storefronts_by_id()   A

Complexity

Conditions 1

Size

Total Lines 6
Code Lines 5

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 4
CRAP Score 1

Importance

Changes 0
Metric Value
cc 1
eloc 5
nop 1
dl 0
loc 6
ccs 4
cts 4
cp 1
crap 1
rs 10
c 0
b 0
f 0
1
"""
2
byceps.blueprints.admin.site.views
3
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
4
5
:Copyright: 2006-2021 Jochen Kupperschmidt
6
:License: Revised BSD (see `LICENSE` file for details)
7
"""
8
9 1
from __future__ import annotations
10 1
import dataclasses
11 1
from typing import Iterable, Iterator
12
13 1
from flask import abort, request
14 1
from flask_babel import gettext
15
16 1
from ....services.board import board_service
17 1
from ....services.brand import service as brand_service
18 1
from ....services.brand.transfer.models import Brand
19 1
from ....services.news import channel_service as news_channel_service
20 1
from ....services.party import service as party_service
21 1
from ....services.shop.shop import service as shop_service
22 1
from ....services.shop.storefront import service as storefront_service
23 1
from ....services.shop.storefront.transfer.models import (
24
    Storefront,
25
    StorefrontID,
26
)
27 1
from ....services.site import (
28
    service as site_service,
29
    settings_service as site_settings_service,
30
)
31 1
from ....services.site.transfer.models import Site, SiteWithBrand
32 1
from ....util.authorization import register_permission_enum
33 1
from ....util.framework.blueprint import create_blueprint
34 1
from ....util.framework.flash import flash_error, flash_success
35 1
from ....util.framework.templating import templated
36 1
from ....util.views import permission_required, redirect_to, respond_no_content
37
38 1
from .authorization import SitePermission
39 1
from .forms import AddNewsChannelForm, CreateForm, UpdateForm
40
41
42 1
blueprint = create_blueprint('site_admin', __name__)
43
44
45 1
register_permission_enum(SitePermission)
46
47
48 1
@blueprint.get('/')
49 1
@permission_required(SitePermission.view)
50 1
@templated
51
def index():
52
    """List all sites."""
53 1
    brands = brand_service.get_all_brands()
54 1
    brands.sort(key=lambda brand: brand.title)
55
56 1
    sites = site_service.get_all_sites()
57 1
    sites = list(_sites_to_sites_with_brand(sites, brands))
58 1
    sites.sort(key=lambda site: (site.title, site.party_id))
59
60 1
    parties = party_service.get_all_parties()
61 1
    party_titles_by_id = {p.id: p.title for p in parties}
62
63 1
    storefronts_by_id = _get_storefronts_by_id(sites)
64
65 1
    return {
66
        'sites': sites,
67
        'brands': brands,
68
        'party_titles_by_id': party_titles_by_id,
69
        'storefronts_by_id': storefronts_by_id,
70
    }
71
72
73 1
@blueprint.get('/for_brand/<brand_id>')
74 1
@permission_required(SitePermission.view)
75 1
@templated
76
def index_for_brand(brand_id):
77
    """List sites for this brand."""
78
    brand = brand_service.find_brand(brand_id)
79
    if brand is None:
80
        abort(404)
81
82
    sites = site_service.get_sites_for_brand(brand.id)
83
    sites = [_site_to_site_with_brand(site, brand) for site in sites]
84
    sites.sort(key=lambda site: (site.title, site.party_id))
85
86
    parties = party_service.get_all_parties()
87
    party_titles_by_id = {p.id: p.title for p in parties}
88
89
    storefronts_by_id = _get_storefronts_by_id(sites)
90
91
    return {
92
        'sites': sites,
93
        'brand': brand,
94
        'party_titles_by_id': party_titles_by_id,
95
        'storefronts_by_id': storefronts_by_id,
96
    }
97
98
99 1
def _sites_to_sites_with_brand(
100
    sites: Iterable[Site], brands: Iterable[Brand]
101
) -> Iterator[SiteWithBrand]:
102 1
    brands_by_id = {brand.id: brand for brand in brands}
103
104 1
    for site in sites:
105 1
        brand = brands_by_id[site.brand_id]
106 1
        yield _site_to_site_with_brand(site, brand)
107
108
109 1
def _site_to_site_with_brand(site: Site, brand: Brand) -> SiteWithBrand:
110 1
    site_tuple = dataclasses.astuple(site)
111 1
    brand_tuple = (brand,)
112
113 1
    return SiteWithBrand(*(site_tuple + brand_tuple))
114
115
116 1
def _get_storefronts_by_id(sites) -> dict[StorefrontID, Storefront]:
117 1
    storefront_ids = {
118
        site.storefront_id for site in sites if site.storefront_id is not None
119
    }
120 1
    storefronts = storefront_service.find_storefronts(storefront_ids)
121 1
    return {storefront.id: storefront for storefront in storefronts}
122
123
124 1
@blueprint.get('/sites/<site_id>')
125 1
@permission_required(SitePermission.view)
126 1
@templated
127
def view(site_id):
128
    """Show a site's settings."""
129 1
    site = site_service.find_site(site_id)
130 1
    if site is None:
131
        abort(404)
132
133 1
    brand = brand_service.find_brand(site.brand_id)
134
135 1
    news_channels = news_channel_service.get_channels(site.news_channel_ids)
136
137 1
    if site.board_id:
138 1
        board = board_service.find_board(site.board_id)
139
    else:
140
        board = None
141
142 1
    if site.storefront_id:
143
        storefront = storefront_service.get_storefront(site.storefront_id)
144
        shop = shop_service.get_shop(storefront.shop_id)
145
    else:
146 1
        storefront = None
147 1
        shop = None
148
149 1
    settings = site_settings_service.get_settings(site.id)
150
151 1
    return {
152
        'site': site,
153
        'brand': brand,
154
        'news_channels': news_channels,
155
        'board': board,
156
        'shop': shop,
157
        'storefront': storefront,
158
        'settings': settings,
159
    }
160
161
162 1
@blueprint.get('/sites/create/for_brand/<brand_id>')
163 1
@permission_required(SitePermission.create)
164 1
@templated
165 1
def create_form(brand_id, erroneous_form=None):
166
    """Show form to create a site."""
167 1
    brand = _get_brand_or_404(brand_id)
168
169 1
    form = erroneous_form if erroneous_form else CreateForm()
170 1
    _fill_in_common_form_choices(form, brand.id)
171
172 1
    return {
173
        'brand': brand,
174
        'form': form,
175
    }
176
177
178 1
@blueprint.post('/sites/for_brand/<brand_id>')
179 1
@permission_required(SitePermission.create)
180
def create(brand_id):
181
    """Create a site."""
182 1
    brand = _get_brand_or_404(brand_id)
183
184 1
    form = CreateForm(request.form)
185 1
    _fill_in_common_form_choices(form, brand.id)
186
187 1
    if not form.validate():
188
        return create_form(brand_id, form)
189
190 1
    site_id = form.id.data.strip().lower()
191 1
    title = form.title.data.strip()
192 1
    server_name = form.server_name.data.strip()
193 1
    party_id = form.party_id.data
194 1
    enabled = form.enabled.data
195 1
    user_account_creation_enabled = form.user_account_creation_enabled.data
196 1
    login_enabled = form.login_enabled.data
197 1
    board_id = form.board_id.data.strip() or None
198 1
    storefront_id = form.storefront_id.data.strip() or None
199
200 1
    if party_id:
201
        party = party_service.find_party(party_id)
202
        if not party:
203
            flash_error(
204
                gettext(
205
                    'Party ID "%(party_id)s" is unknown.',
206
                    party_id=party_id,
207
                )
208
            )
209
            return create_form(brand_id, form)
210
    else:
211 1
        party_id = None
212
213 1
    site = site_service.create_site(
214
        site_id,
215
        title,
216
        server_name,
217
        brand.id,
218
        enabled=enabled,
219
        user_account_creation_enabled=user_account_creation_enabled,
220
        login_enabled=login_enabled,
221
        party_id=party_id,
222
        board_id=board_id,
223
        storefront_id=storefront_id,
224
    )
225
226 1
    flash_success(
227
        gettext('Site "%(title)s" has been created.', title=site.title)
228
    )
229
230 1
    return redirect_to('.view', site_id=site.id)
231
232
233 1
@blueprint.get('/sites/<site_id>/update')
234 1
@permission_required(SitePermission.update)
235 1
@templated
236 1
def update_form(site_id, erroneous_form=None):
237
    """Show form to update the site."""
238 1
    site = _get_site_or_404(site_id)
239
240 1
    form = erroneous_form if erroneous_form else UpdateForm(obj=site)
241 1
    form.set_brand_choices()
242 1
    _fill_in_common_form_choices(form, site.brand_id)
243
244 1
    return {
245
        'site': site,
246
        'form': form,
247
    }
248
249
250 1
@blueprint.post('/sites/<site_id>')
251 1
@permission_required(SitePermission.update)
252
def update(site_id):
253
    """Update the site."""
254
    site = _get_site_or_404(site_id)
255
256
    form = UpdateForm(request.form)
257
    form.set_brand_choices()
258
    _fill_in_common_form_choices(form, site.brand_id)
259
260
    if not form.validate():
261
        return update_form(site.id, form)
262
263
    title = form.title.data.strip()
264
    server_name = form.server_name.data.strip()
265
    brand_id = form.brand_id.data
266
    party_id = form.party_id.data
267
    enabled = form.enabled.data
268
    user_account_creation_enabled = form.user_account_creation_enabled.data
269
    login_enabled = form.login_enabled.data
270
    board_id = form.board_id.data.strip() or None
271
    storefront_id = form.storefront_id.data.strip() or None
272
    archived = form.archived.data
273
274
    if party_id:
275
        party = party_service.find_party(party_id)
276
        if not party:
277
            flash_error(
278
                gettext(
279
                    'Party ID "%(party_id)s" is unknown.',
280
                    party_id=party_id,
281
                )
282
            )
283
            return update_form(site.id, form)
284
    else:
285
        party_id = None
286
287
    try:
288
        site = site_service.update_site(
289
            site.id,
290
            title,
291
            server_name,
292
            brand_id,
293
            party_id,
294
            enabled,
295
            user_account_creation_enabled,
296
            login_enabled,
297
            board_id,
298
            storefront_id,
299
            archived,
300
        )
301
    except site_service.UnknownSiteId:
302
        abort(404, f'Unknown site ID "{site_id}".')
303
304
    flash_success(
305
        gettext('Site "%(title)s" has been updated.', title=site.title)
306
    )
307
308
    return redirect_to('.view', site_id=site.id)
309
310
311 1
def _fill_in_common_form_choices(form, brand_id):
312 1
    form.set_party_choices(brand_id)
313 1
    form.set_board_choices(brand_id)
314 1
    form.set_storefront_choices()
315
316
317
# -------------------------------------------------------------------- #
318
# news channels
319
320
321 1
@blueprint.get('/sites/<site_id>/news_channels/add')
322 1
@permission_required(SitePermission.update)
323 1
@templated
324 1
def add_news_channel_form(site_id, erroneous_form=None):
325
    """Show form to add a news channel to the site."""
326
    site = _get_site_or_404(site_id)
327
328
    form = erroneous_form if erroneous_form else AddNewsChannelForm()
329
    form.set_news_channel_choices(site.brand_id)
330
331
    return {
332
        'site': site,
333
        'form': form,
334
    }
335
336
337 1
@blueprint.post('/sites/<site_id>/news_channels')
338 1
@permission_required(SitePermission.update)
339
def add_news_channel(site_id):
340
    """Add a news channel to the site."""
341
    site = _get_site_or_404(site_id)
342
343
    form = AddNewsChannelForm(request.form)
344
    form.set_news_channel_choices(site.brand_id)
345
346
    if not form.validate():
347
        return add_news_channel_form(site.id, form)
348
349
    news_channel_id = form.news_channel_id.data
350
    news_channel = news_channel_service.get_channel(news_channel_id)
351
352
    site_service.add_news_channel(site.id, news_channel.id)
353
354
    flash_success(
355
        gettext(
356
            'News channel "%(news_channel_id)s" has been added to site "%(site_title)s".',
357
            news_channel_id=news_channel.id,
358
            site_title=site.title,
359
        )
360
    )
361
362
    return redirect_to('.view', site_id=site.id)
363
364
365 1
@blueprint.delete('/sites/<site_id>/news_channels/<news_channel_id>')
366 1
@permission_required(SitePermission.update)
367 1
@respond_no_content
368
def remove_news_channel(site_id, news_channel_id):
369
    """Remove the news channel from the site."""
370
    site = _get_site_or_404(site_id)
371
372
    news_channel = news_channel_service.find_channel(news_channel_id)
373
    if news_channel is None:
374
        abort(404)
375
376
    news_channel_id = news_channel.id
377
378
    site_service.remove_news_channel(site.id, news_channel.id)
379
380
    flash_success(
381
        gettext(
382
            'News channel "%(news_channel_id)s" has been removed from site "%(site_title)s".',
383
            news_channel_id=news_channel.id,
384
            site_title=site.title,
385
        )
386
    )
387
388
389
# -------------------------------------------------------------------- #
390
391
392 1
def _get_site_or_404(site_id):
393 1
    site = site_service.find_site(site_id)
394
395 1
    if site is None:
396
        abort(404)
397
398 1
    return site
399
400
401 1
def _get_brand_or_404(brand_id):
402 1
    brand = brand_service.find_brand(brand_id)
403
404 1
    if brand is None:
405
        abort(404)
406
407
    return brand
408