~homeworkprod/byceps

ref: 533662eefb314d54b168e5a46bc07c88b5f88be4 byceps/byceps/blueprints/admin/newsletter/views.py -rw-r--r-- 3.1 KiB
533662ee — Jochen Kupperschmidt Use Flask v2.0's `delete`/`get`/`patch`/`post` route shortcuts 2 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
"""
byceps.blueprints.admin.newsletter.views
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

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

from dataclasses import dataclass
from operator import attrgetter

from flask import abort

from ....services.newsletter import service as newsletter_service
from ....services.newsletter.transfer.models import List
from ....services.newsletter.types import SubscriptionState
from ....util.authorization import register_permission_enum
from ....util.framework.blueprint import create_blueprint
from ....util.framework.templating import templated
from ....util.views import jsonified, permission_required, textified

from .authorization import NewsletterPermission


blueprint = create_blueprint('newsletter_admin', __name__)


register_permission_enum(NewsletterPermission)


@dataclass(frozen=True)
class ListWithStats(List):
    subscriber_count: int


@blueprint.get('/lists')
@permission_required(NewsletterPermission.view_subscriptions)
@templated
def index():
    """List all lists."""
    lists = newsletter_service.get_all_lists()

    lists_with_stats = list(map(_add_subscriber_count, lists))

    return {
        'lists': lists_with_stats,
    }


def _add_subscriber_count(list_):
    subscriber_count = newsletter_service.count_subscribers_for_list(list_.id)

    return ListWithStats(list_.id, list_.title, subscriber_count)


@blueprint.get('/lists/<list_id>/subscriptions')
@permission_required(NewsletterPermission.view_subscriptions)
@templated
def view_subscriptions(list_id):
    """Show user subscription states for that list."""
    list_ = _get_list_or_404(list_id)

    totals = newsletter_service.count_subscriptions_by_state(list_.id)

    return {
        'list_': list_,
        'totals': totals,
        'State': SubscriptionState,
    }


@blueprint.get('/lists/<list_id>/subscriptions/export')
@permission_required(NewsletterPermission.export_subscribers)
@jsonified
def export_subscribers(list_id):
    """Export the screen names and email addresses of enabled users
    which are currently subscribed to that list as JSON.
    """
    list_ = _get_list_or_404(list_id)

    subscribers = newsletter_service.get_subscribers(list_.id)

    exports = list(map(assemble_subscriber_export, subscribers))

    return {'subscribers': exports}


def assemble_subscriber_export(subscriber):
    return {
        'screen_name': subscriber.screen_name,
        'email_address': subscriber.email_address,
    }


@blueprint.get('/lists/<list_id>/subscriptions/email_addresses/export')
@permission_required(NewsletterPermission.export_subscribers)
@textified
def export_subscriber_email_addresses(list_id):
    """Export the email addresses of enabled users which are currently
    subscribed to that list as plaintext, with one address per row.
    """
    list_ = _get_list_or_404(list_id)

    subscribers = newsletter_service.get_subscribers(list_.id)
    email_addresses = map(attrgetter('email_address'), subscribers)
    return '\n'.join(email_addresses)


def _get_list_or_404(list_id):
    list_ = newsletter_service.find_list(list_id)

    if list_ is None:
        abort(404)

    return list_