byceps.blueprints.site.board.views_topic   A
last analyzed

Complexity

Total Complexity 40

Size/Duplication

Total Lines 526
Duplicated Lines 0 %

Test Coverage

Coverage 69.03%

Importance

Changes 0
Metric Value
eloc 319
dl 0
loc 526
ccs 156
cts 226
cp 0.6903
rs 9.2
c 0
b 0
f 0
wmc 40

17 Functions

Rating   Name   Duplication   Size   Complexity  
A topic_create_form() 0 13 2
A topic_index() 0 22 1
A topic_moderate_form() 0 17 1
B topic_update_form() 0 34 6
B topic_update() 0 35 6
A topic_unhide() 0 24 1
A topic_pin() 0 21 1
A topic_unpin() 0 20 1
A topic_lock() 0 21 1
C topic_view() 0 78 8
A topic_remove_limit_to_announcements() 0 17 1
A topic_move() 0 36 2
A topic_limit_to_announcements() 0 18 1
A topic_hide() 0 21 1
A _find_posting_url_to_redirect_to() 0 26 4
A topic_create() 0 27 2
A topic_unlock() 0 21 1

How to fix   Complexity   

Complexity

Complex classes like byceps.blueprints.site.board.views_topic often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.

Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.

1
"""
2
byceps.blueprints.site.board.views_topic
3
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
4
5
:Copyright: 2006-2021 Jochen Kupperschmidt
6
:License: Revised BSD (see `LICENSE` file for details)
7
"""
8
9 1
import dataclasses
10 1
from datetime import datetime
11 1
from typing import Optional
12
13 1
from flask import abort, g, redirect, request
14 1
from flask_babel import gettext
15
16 1
from ....services.authentication.session.models.current_user import CurrentUser
17 1
from ....services.board import (
18
    category_query_service as board_category_query_service,
19
    last_view_service as board_last_view_service,
20
    posting_query_service as board_posting_query_service,
21
    topic_command_service as board_topic_command_service,
22
    topic_query_service as board_topic_query_service,
23
)
24 1
from ....services.board.transfer.models import TopicID
25 1
from ....services.text_markup.service import get_smileys
26 1
from ....services.user import service as user_service
27 1
from ....signals import board as board_signals
28 1
from ....util.framework.flash import flash_error, flash_success
29 1
from ....util.framework.templating import templated
30 1
from ....util.views import permission_required, respond_no_content_with_location
31
32 1
from ..orga_team.service import is_orga_for_current_party
33
34 1
from .blueprint import blueprint
35 1
from .forms import PostingCreateForm, TopicCreateForm, TopicUpdateForm
36 1
from . import _helpers as h, service
37
38
39 1
@blueprint.get('/topics', defaults={'page': 1})
40 1
@blueprint.get('/topics/pages/<int:page>')
41 1
@templated
42
def topic_index(page):
43
    """List latest topics in all categories."""
44 1
    board_id = h.get_board_id()
45 1
    user = g.user
46
47 1
    h.require_board_access(board_id, user.id)
48
49 1
    include_hidden = service.may_current_user_view_hidden()
50 1
    topics_per_page = service.get_topics_per_page_value()
51
52 1
    topics = board_topic_query_service.paginate_topics(
53
        board_id, include_hidden, page, topics_per_page
54
    )
55
56 1
    service.add_topic_creators(topics.items)
57 1
    service.add_topic_unseen_flag(topics.items, user)
58
59 1
    return {
60
        'topics': topics,
61
    }
62
63
64 1
@blueprint.get('/topics/<uuid:topic_id>', defaults={'page': 0})
65 1
@blueprint.get('/topics/<uuid:topic_id>/pages/<int:page>')
66 1
@templated
67
def topic_view(topic_id, page):
68
    """List postings for the topic."""
69 1
    user = g.user
70
71 1
    include_hidden = service.may_current_user_view_hidden()
72
73 1
    topic = board_topic_query_service.find_topic_visible_for_user(
74
        topic_id, include_hidden
75
    )
76
77 1
    if topic is None:
78
        abort(404)
79
80 1
    board_id = h.get_board_id()
81
82 1
    if topic.category.hidden or (topic.category.board_id != board_id):
83
        abort(404)
84
85 1
    h.require_board_access(board_id, user.id)
86
87
    # Copy last view timestamp for later use to compare postings
88
    # against it.
89 1
    last_viewed_at = board_last_view_service.find_topic_last_viewed_at(
90
        topic.id, user.id
91
    )
92
93 1
    postings_per_page = service.get_postings_per_page_value()
94 1
    if page == 0:
95 1
        posting_url_to_redirect_to = _find_posting_url_to_redirect_to(
96
            topic.id, user, include_hidden, last_viewed_at
97
        )
98
99 1
        if posting_url_to_redirect_to is not None:
100
            # Jump to a specific post. This requires a redirect.
101
            return redirect(posting_url_to_redirect_to, code=307)
102
103 1
        page = 1
104
105 1
    if user.authenticated:
106
        # Mark as viewed before aborting so a user can itself remove the
107
        # 'new' tag from a locked topic.
108 1
        board_last_view_service.mark_topic_as_just_viewed(topic.id, user.id)
109
110 1
    postings = board_posting_query_service.paginate_postings(
111
        topic.id, include_hidden, page, postings_per_page
112
    )
113
114 1
    service.add_unseen_flag_to_postings(postings.items, last_viewed_at)
115
116 1
    is_last_page = not postings.has_next
117
118 1
    service.enrich_creators(postings.items, g.brand_id, g.party_id)
119
120 1
    is_current_user_orga = user.authenticated and is_orga_for_current_party(
121
        g.user.id
122
    )
123
124 1
    context = {
125
        'topic': topic,
126
        'postings': postings,
127
        'is_last_page': is_last_page,
128
        'may_topic_be_updated_by_current_user': service.may_topic_be_updated_by_current_user,
129
        'may_posting_be_updated_by_current_user': service.may_posting_be_updated_by_current_user,
130
        'is_current_user_orga': is_current_user_orga,
131
    }
132
133 1
    if is_last_page:
134 1
        context.update(
135
            {
136
                'form': PostingCreateForm(),
137
                'smileys': get_smileys(),
138
            }
139
        )
140
141 1
    return context
142
143
144 1
def _find_posting_url_to_redirect_to(
145
    topic_id: TopicID,
146
    user: CurrentUser,
147
    include_hidden: bool,
148
    last_viewed_at: Optional[datetime],
149
) -> Optional[str]:
150 1
    if not user.authenticated:
151
        # All postings are potentially new to a guest, so start on
152
        # the first page.
153 1
        return None
154
155 1
    if last_viewed_at is None:
156
        # This topic is completely new to the current user, so
157
        # start on the first page.
158 1
        return None
159
160
    posting = board_topic_query_service.find_default_posting_to_jump_to(
161
        topic_id, include_hidden, last_viewed_at
162
    )
163
164
    if posting is None:
165
        return None
166
167
    page = service.calculate_posting_page_number(posting)
168
169
    return h.build_url_for_posting_in_topic_view(posting, page)
170
171
172 1
@blueprint.get('/categories/<category_id>/create')
173 1
@permission_required('board_topic.create')
174 1
@templated
175 1
def topic_create_form(category_id, erroneous_form=None):
176
    """Show a form to create a topic in the category."""
177
    category = h.get_category_or_404(category_id)
178
179
    form = erroneous_form if erroneous_form else TopicCreateForm()
180
181
    return {
182
        'category': category,
183
        'form': form,
184
        'smileys': get_smileys(),
185
    }
186
187
188 1
@blueprint.post('/categories/<category_id>/create')
189 1
@permission_required('board_topic.create')
190
def topic_create(category_id):
191
    """Create a topic in the category."""
192
    category = h.get_category_or_404(category_id)
193
194
    form = TopicCreateForm(request.form)
195
    if not form.validate():
196
        return topic_create_form(category.id, form)
197
198
    creator = g.user
199
    title = form.title.data.strip()
200
    body = form.body.data.strip()
201
202
    topic, event = board_topic_command_service.create_topic(
203
        category.id, creator.id, title, body
204
    )
205
    topic_url = h.build_external_url_for_topic(topic.id)
206
207
    flash_success(
208
        gettext('Topic "%(title)s" has been created.', title=topic.title)
209
    )
210
211
    event = dataclasses.replace(event, url=topic_url)
212
    board_signals.topic_created.send(None, event=event)
213
214
    return redirect(topic_url)
215
216
217 1
@blueprint.get('/topics/<uuid:topic_id>/update')
218 1
@permission_required('board_topic.update')
219 1
@templated
220 1
def topic_update_form(topic_id, erroneous_form=None):
221
    """Show form to update a topic."""
222
    topic = h.get_topic_or_404(topic_id)
223
    url = h.build_url_for_topic(topic.id)
224
225
    user_may_update = service.may_topic_be_updated_by_current_user(topic)
226
227
    if topic.locked and not user_may_update:
228
        flash_error(
229
            gettext('The topic must not be updated because it is locked.')
230
        )
231
        return redirect(url)
232
233
    if topic.hidden:
234
        flash_error(gettext('The topic must not be updated.'))
235
        return redirect(url)
236
237
    if not user_may_update:
238
        flash_error(gettext('You are not allowed to update this topic.'))
239
        return redirect(url)
240
241
    form = (
242
        erroneous_form
243
        if erroneous_form
244
        else TopicUpdateForm(obj=topic, body=topic.initial_posting.body)
245
    )
246
247
    return {
248
        'form': form,
249
        'topic': topic,
250
        'smileys': get_smileys(),
251
    }
252
253
254 1
@blueprint.post('/topics/<uuid:topic_id>')
255 1
@permission_required('board_topic.update')
256
def topic_update(topic_id):
257
    """Update a topic."""
258
    topic = h.get_topic_or_404(topic_id)
259
    url = h.build_url_for_topic(topic.id)
260
261
    user_may_update = service.may_topic_be_updated_by_current_user(topic)
262
263
    if topic.locked and not user_may_update:
264
        flash_error(
265
            gettext('The topic must not be updated because it is locked.')
266
        )
267
        return redirect(url)
268
269
    if topic.hidden:
270
        flash_error(gettext('The topic must not be updated.'))
271
        return redirect(url)
272
273
    if not user_may_update:
274
        flash_error(gettext('You are not allowed to update this topic.'))
275
        return redirect(url)
276
277
    form = TopicUpdateForm(request.form)
278
    if not form.validate():
279
        return topic_update_form(topic_id, form)
280
281
    board_topic_command_service.update_topic(
282
        topic.id, g.user.id, form.title.data, form.body.data
283
    )
284
285
    flash_success(
286
        gettext('Topic "%(title)s" has been updated.', title=topic.title)
287
    )
288
    return redirect(url)
289
290
291 1
@blueprint.get('/topics/<uuid:topic_id>/moderate')
292 1
@permission_required('board.hide')
293 1
@templated
294
def topic_moderate_form(topic_id):
295
    """Show a form to moderate the topic."""
296
    board_id = h.get_board_id()
297
    topic = h.get_topic_or_404(topic_id)
298
299
    topic.creator = user_service.get_user(topic.creator_id)
300
301
    categories = board_category_query_service.get_categories_excluding(
302
        board_id, topic.category_id
303
    )
304
305
    return {
306
        'topic': topic,
307
        'categories': categories,
308
    }
309
310
311 1
@blueprint.post('/topics/<uuid:topic_id>/flags/hidden')
312 1
@permission_required('board.hide')
313 1
@respond_no_content_with_location
314
def topic_hide(topic_id):
315
    """Hide a topic."""
316 1
    topic = h.get_topic_or_404(topic_id)
317 1
    moderator_id = g.user.id
318
319 1
    event = board_topic_command_service.hide_topic(topic.id, moderator_id)
320
321 1
    flash_success(
322
        gettext('Topic "%(title)s" has been hidden.', title=topic.title),
323
        icon='hidden',
324
    )
325
326 1
    event = dataclasses.replace(
327
        event, url=h.build_external_url_for_topic(topic.id)
328
    )
329 1
    board_signals.topic_hidden.send(None, event=event)
330
331 1
    return h.build_url_for_topic_in_category_view(topic)
332
333
334 1
@blueprint.delete('/topics/<uuid:topic_id>/flags/hidden')
335 1
@permission_required('board.hide')
336 1
@respond_no_content_with_location
337
def topic_unhide(topic_id):
338
    """Un-hide a topic."""
339 1
    topic = h.get_topic_or_404(topic_id)
340 1
    moderator_id = g.user.id
341
342 1
    event = board_topic_command_service.unhide_topic(topic.id, moderator_id)
343
344 1
    flash_success(
345
        gettext(
346
            'Topic "%(title)s" has been made visible again.',
347
            title=topic.title,
348
        ),
349
        icon='view',
350
    )
351
352 1
    event = dataclasses.replace(
353
        event, url=h.build_external_url_for_topic(topic.id)
354
    )
355 1
    board_signals.topic_unhidden.send(None, event=event)
356
357 1
    return h.build_url_for_topic_in_category_view(topic)
358
359
360 1
@blueprint.post('/topics/<uuid:topic_id>/flags/locked')
361 1
@permission_required('board_topic.lock')
362 1
@respond_no_content_with_location
363
def topic_lock(topic_id):
364
    """Lock a topic."""
365 1
    topic = h.get_topic_or_404(topic_id)
366 1
    moderator_id = g.user.id
367
368 1
    event = board_topic_command_service.lock_topic(topic.id, moderator_id)
369
370 1
    flash_success(
371
        gettext('Topic "%(title)s" has been locked.', title=topic.title),
372
        icon='lock',
373
    )
374
375 1
    event = dataclasses.replace(
376
        event, url=h.build_external_url_for_topic(topic.id)
377
    )
378 1
    board_signals.topic_locked.send(None, event=event)
379
380 1
    return h.build_url_for_topic_in_category_view(topic)
381
382
383 1
@blueprint.delete('/topics/<uuid:topic_id>/flags/locked')
384 1
@permission_required('board_topic.lock')
385 1
@respond_no_content_with_location
386
def topic_unlock(topic_id):
387
    """Unlock a topic."""
388 1
    topic = h.get_topic_or_404(topic_id)
389 1
    moderator_id = g.user.id
390
391 1
    event = board_topic_command_service.unlock_topic(topic.id, moderator_id)
392
393 1
    flash_success(
394
        gettext('Topic "%(title)s" has been unlocked.', title=topic.title),
395
        icon='unlock',
396
    )
397
398 1
    event = dataclasses.replace(
399
        event, url=h.build_external_url_for_topic(topic.id)
400
    )
401 1
    board_signals.topic_unlocked.send(None, event=event)
402
403 1
    return h.build_url_for_topic_in_category_view(topic)
404
405
406 1
@blueprint.post('/topics/<uuid:topic_id>/flags/pinned')
407 1
@permission_required('board_topic.pin')
408 1
@respond_no_content_with_location
409
def topic_pin(topic_id):
410
    """Pin a topic."""
411 1
    topic = h.get_topic_or_404(topic_id)
412 1
    moderator_id = g.user.id
413
414 1
    event = board_topic_command_service.pin_topic(topic.id, moderator_id)
415
416 1
    flash_success(
417
        gettext('Topic "%(title)s" has been pinned.', title=topic.title),
418
        icon='pin',
419
    )
420
421 1
    event = dataclasses.replace(
422
        event, url=h.build_external_url_for_topic(topic.id)
423
    )
424 1
    board_signals.topic_pinned.send(None, event=event)
425
426 1
    return h.build_url_for_topic_in_category_view(topic)
427
428
429 1
@blueprint.delete('/topics/<uuid:topic_id>/flags/pinned')
430 1
@permission_required('board_topic.pin')
431 1
@respond_no_content_with_location
432
def topic_unpin(topic_id):
433
    """Unpin a topic."""
434 1
    topic = h.get_topic_or_404(topic_id)
435 1
    moderator_id = g.user.id
436
437 1
    event = board_topic_command_service.unpin_topic(topic.id, moderator_id)
438
439 1
    flash_success(
440
        gettext('Topic "%(title)s" has been unpinned.', title=topic.title)
441
    )
442
443 1
    event = dataclasses.replace(
444
        event, url=h.build_external_url_for_topic(topic.id)
445
    )
446 1
    board_signals.topic_unpinned.send(None, event=event)
447
448 1
    return h.build_url_for_topic_in_category_view(topic)
449
450
451 1
@blueprint.post('/topics/<uuid:topic_id>/move')
452 1
@permission_required('board_topic.move')
453
def topic_move(topic_id):
454
    """Move a topic from one category to another."""
455 1
    topic = h.get_topic_or_404(topic_id)
456 1
    moderator_id = g.user.id
457
458 1
    new_category_id = request.form.get('category_id')
459 1
    if not new_category_id:
460
        abort(400, 'No target category ID given.')
461
462 1
    new_category = h.get_category_or_404(new_category_id)
463
464 1
    old_category = topic.category
465
466 1
    event = board_topic_command_service.move_topic(
467
        topic.id, new_category.id, moderator_id
468
    )
469
470 1
    flash_success(
471
        gettext(
472
            'Topic "%(topic_title)s" has been moved from category '
473
            '"%(old_category_title)s" to category "%(new_category_title)s".',
474
            topic_title=topic.title,
475
            old_category_title=old_category.title,
476
            new_category_title=new_category.title,
477
        ),
478
        icon='move',
479
    )
480
481 1
    event = dataclasses.replace(
482
        event, url=h.build_external_url_for_topic(topic.id)
483
    )
484 1
    board_signals.topic_moved.send(None, event=event)
485
486 1
    return redirect(h.build_url_for_topic_in_category_view(topic))
487
488
489 1
@blueprint.post('/topics/<uuid:topic_id>/flags/announcements')
490 1
@permission_required('board.announce')
491 1
@respond_no_content_with_location
492
def topic_limit_to_announcements(topic_id):
493
    """Limit post in the topic to moderators."""
494
    topic = h.get_topic_or_404(topic_id)
495
496
    board_topic_command_service.limit_topic_to_announcements(topic.id)
497
498
    flash_success(
499
        gettext(
500
            'Topic "%(title)s" has been limited to announcements.',
501
            title=topic.title,
502
        ),
503
        icon='announce',
504
    )
505
506
    return h.build_url_for_topic_in_category_view(topic)
507
508
509 1
@blueprint.delete('/topics/<uuid:topic_id>/flags/announcements')
510 1
@permission_required('board.announce')
511 1
@respond_no_content_with_location
512
def topic_remove_limit_to_announcements(topic_id):
513
    """Allow non-moderators to post in the topic again."""
514
    topic = h.get_topic_or_404(topic_id)
515
516
    board_topic_command_service.remove_limit_of_topic_to_announcements(topic.id)
517
518
    flash_success(
519
        gettext(
520
            'Topic "%(title)s" has been reopened for regular replies.',
521
            title=topic.title,
522
        )
523
    )
524
525
    return h.build_url_for_topic_in_category_view(topic)
526