byceps.services.newsletter.service._db_entity_to_list() - Code Metrics - Inspection of "Introduce brand-independent newsletter lists" - byceps/byceps - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 97b1b2...a0bd49 )

by Jochen

created 2019-08-17 14:41 UTC

byceps.services.newsletter.service._db_entity_to_list() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	4
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	4
nop	1
dl	0
loc	4
rs	10
c	0
b	0
f	0

"""
byceps.services.newsletter.service
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

:Copyright: 2006-2019 Jochen Kupperschmidt
:License: Modified BSD, see LICENSE for details.
"""

from collections import Counter
from operator import itemgetter
from typing import Any, Dict, Iterable, Iterator, Optional, Sequence, Set, \
    Tuple, Union

from ...database import BaseQuery, db
from ...typing import UserID

from ..user.models.user import User as DbUser
from ..user.transfer.models import User

from .models import List as DbList, Subscriber, Subscription
from .transfer.models import List, ListID
from .types import SubscriptionState


def find_list(list_id: ListID) -> Optional[List]:
    """Return the list with that ID, or `None` if not found."""
    list_ = DbList.query.get(list_id)

    if list_ is None:
        return None

    return _db_entity_to_list(list_)


def get_all_lists() -> Sequence[List]:
    """Return all lists."""
    lists = DbList.query.all()

    return [_db_entity_to_list(list_) for list_ in lists]


def count_subscribers_for_list(list_id: ListID) -> int:
    """Return the number of users that are currently subscribed to that list."""
    return _build_query_for_current_subscribers(list_id).count()


def get_subscribers(list_id: ListID) -> Iterable[Subscriber]:
    """Yield screen name and email address of the enabled users that
    are currently subscribed to the list.
    """
    subscriber_id_rows = _build_query_for_current_subscribers(list_id).all()

    subscriber_ids = set(map(itemgetter(0), subscriber_id_rows))

    return _get_subscriber_details(subscriber_ids)


def _build_query_for_current_subscribers(list_id: ListID) -> BaseQuery:
    """Build a query to return the most recent subscription state
    (grouped by user and list).

    The generated SQL should be equivalent to this:

        SELECT
          nso.user_id
        FROM newsletter_subscriptions AS nso
          JOIN (
            SELECT
              user_id,
              list_id,
              MAX(expressed_at) AS latest_expressed_at
            FROM newsletter_subscriptions
            GROUP BY
              user_id,
              list_id
          ) AS nsi
            ON nso.user_id = nsi.user_id
              AND nso.list_id = nsi.list_id
              AND nso.expressed_at = nsi.latest_expressed_at
        WHERE nso.state = 'requested'
          AND nso.list_id = <list_id>
    """
    subquery = _build_query_for_latest_expressed_at().subquery()

    return db.session \
        .query(
            Subscription.user_id
        ) \
        .join(subquery, db.and_(
            Subscription.user_id == subquery.c.user_id,
            Subscription.list_id == subquery.c.list_id,
            Subscription.expressed_at == subquery.c.latest_expressed_at
        )) \
        .filter(Subscription._state == SubscriptionState.requested.name) \
        .filter(Subscription.list_id == list_id)


def _get_subscriber_details(user_ids: Set[UserID]) -> Iterator[Subscriber]:
    """Yield screen name and email address of each user (if enabled)."""
    if not user_ids:
        return []

    rows = db.session \
        .query(
            DbUser.screen_name,
            DbUser.email_address,
        ) \
        .filter(DbUser.id.in_(user_ids)) \
        .filter_by(enabled=True) \
        .filter_by(suspended=False) \
        .filter_by(deleted=False) \
        .all()

    for row in rows:
        yield Subscriber(row.screen_name, row.email_address)


def count_subscriptions_by_state(list_id: ListID
                                ) -> Dict[Union[SubscriptionState, str], int]:
    """Return the totals for each state as well as an overall total."""
    rows = _build_query_for_current_state(list_id) \
        .all()

    totals = {state: 0 for state in SubscriptionState}

    for state_name, count in rows:
        state = SubscriptionState[state_name]
        totals[state] = count

    totals['total'] = sum(totals.values())

    return totals


def _build_query_for_current_state(list_id: ListID) -> BaseQuery:
    """Build a query to return the number of currently requested and
    declined subscription states for that list.

    The generated SQL should be equivalent to this:

        SELECT
          nso.state,
          COUNT(nso.state)
        FROM newsletter_subscriptions AS nso
          JOIN (
            SELECT
              user_id,
              list_id,
              MAX(expressed_at) AS latest_expressed_at
            FROM newsletter_subscriptions
            GROUP BY
              user_id,
              list_id
          ) AS nsi
            ON nso.user_id = nsi.user_id
              AND nso.list_id = nsi.list_id
              AND nso.expressed_at = nsi.latest_expressed_at
        WHERE list_id = {list_id}
        GROUP BY
          list_id,
          state
    """
    subquery = _build_query_for_latest_expressed_at().subquery()

    return db.session \
        .query(
            Subscription._state,
            db.func.count(Subscription._state),
        ) \
        .join(subquery, db.and_(
            Subscription.user_id == subquery.c.user_id,
            Subscription.list_id == subquery.c.list_id,
            Subscription.expressed_at == subquery.c.latest_expressed_at
        )) \
        .filter_by(list_id=list_id) \
        .group_by(
            Subscription.list_id,
            Subscription._state,
        )


def _build_query_for_latest_expressed_at() -> BaseQuery:
    """Build a query to return the most recent time the subscription
    state was set (grouped by user and list).

    The generated SQL should be equivalent to this:

        SELECT user_id, list_id, MAX(expressed_at) AS latest_expressed_at
        FROM newsletter_subscriptions
        GROUP BY user_id, list_id
    """
    return db.session \
        .query(
            Subscription.user_id,
            Subscription.list_id,
            db.func.max(Subscription.expressed_at).label('latest_expressed_at')
        ) \
        .group_by(
            Subscription.user_id,
            Subscription.list_id
        )


def get_subscription_state(user_id: UserID, list_id: ListID
                          ) -> SubscriptionState:
    """Return the user's current subscription state for that list."""
    current_subscription = Subscription.query \
        .filter_by(user_id=user_id) \
        .filter_by(list_id=list_id) \
        .order_by(Subscription.expressed_at.desc()) \
        .first()

    if current_subscription is None:
        return SubscriptionState.declined

    return current_subscription.state


def get_subscription_updates_for_user(user_id: UserID
                                     ) -> Sequence[Subscription]:
    """Return subscription updates made by the user, for any list."""
    return Subscription.query \
        .filter_by(user_id=user_id) \
        .all()


def is_subscribed(user_id: UserID, list_id: ListID) -> bool:
    """Return if the user is subscribed to the list or not."""
    subscription_state = get_subscription_state(user_id, list_id)
    return subscription_state == SubscriptionState.requested


def _db_entity_to_list(list_: DbList) -> List:
    return List(
        list_.id,
        list_.title,
    )


1			"""
2			byceps.services.newsletter.service
3			~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
4
5			:Copyright: 2006-2019 Jochen Kupperschmidt
6			:License: Modified BSD, see LICENSE for details.
7			"""
8
9			from collections import Counter
10			from operator import itemgetter
11			from typing import Any, Dict, Iterable, Iterator, Optional, Sequence, Set, \
12			Tuple, Union
13
14			from ...database import BaseQuery, db
15			from ...typing import UserID
16
17			from ..user.models.user import User as DbUser
18			from ..user.transfer.models import User
19
20			from .models import List as DbList, Subscriber, Subscription
21			from .transfer.models import List, ListID
22			from .types import SubscriptionState
23
24
25			def find_list(list_id: ListID) -> Optional[List]:
26			"""Return the list with that ID, or `None` if not found."""
27			list_ = DbList.query.get(list_id)
28
29			if list_ is None:
30			return None
31
32			return _db_entity_to_list(list_)
33
34
35			def get_all_lists() -> Sequence[List]:
36			"""Return all lists."""
37			lists = DbList.query.all()
38
39			return [_db_entity_to_list(list_) for list_ in lists]
40
41
42			def count_subscribers_for_list(list_id: ListID) -> int:
43			"""Return the number of users that are currently subscribed to that list."""
44			return _build_query_for_current_subscribers(list_id).count()
45
46
47			def get_subscribers(list_id: ListID) -> Iterable[Subscriber]:
48			"""Yield screen name and email address of the enabled users that
49			are currently subscribed to the list.
50			"""
51			subscriber_id_rows = _build_query_for_current_subscribers(list_id).all()
52
53			subscriber_ids = set(map(itemgetter(0), subscriber_id_rows))
54
55			return _get_subscriber_details(subscriber_ids)
56
57
58			def _build_query_for_current_subscribers(list_id: ListID) -> BaseQuery:
59			"""Build a query to return the most recent subscription state
60			(grouped by user and list).
61
62			The generated SQL should be equivalent to this:
63
64			SELECT
65			nso.user_id
66			FROM newsletter_subscriptions AS nso
67			JOIN (
68			SELECT
69			user_id,
70			list_id,
71			MAX(expressed_at) AS latest_expressed_at
72			FROM newsletter_subscriptions
73			GROUP BY
74			user_id,
75			list_id
76			) AS nsi
77			ON nso.user_id = nsi.user_id
78			AND nso.list_id = nsi.list_id
79			AND nso.expressed_at = nsi.latest_expressed_at
80			WHERE nso.state = 'requested'
81			AND nso.list_id = <list_id>
82			"""
83			subquery = _build_query_for_latest_expressed_at().subquery()
84
85			return db.session \
86			.query(
87			Subscription.user_id
88			) \
89			.join(subquery, db.and_(
90			Subscription.user_id == subquery.c.user_id,
91			Subscription.list_id == subquery.c.list_id,
92			Subscription.expressed_at == subquery.c.latest_expressed_at
93			)) \
94			.filter(Subscription._state == SubscriptionState.requested.name) \
95			.filter(Subscription.list_id == list_id)
96
97
98			def _get_subscriber_details(user_ids: Set[UserID]) -> Iterator[Subscriber]:
99			"""Yield screen name and email address of each user (if enabled)."""
100			if not user_ids:
101			return []
102
103			rows = db.session \
104			.query(
105			DbUser.screen_name,
106			DbUser.email_address,
107			) \
108			.filter(DbUser.id.in_(user_ids)) \
109			.filter_by(enabled=True) \
110			.filter_by(suspended=False) \
111			.filter_by(deleted=False) \
112			.all()
113
114			for row in rows:
115			yield Subscriber(row.screen_name, row.email_address)
116
117
118			def count_subscriptions_by_state(list_id: ListID
119			) -> Dict[Union[SubscriptionState, str], int]:
120			"""Return the totals for each state as well as an overall total."""
121			rows = _build_query_for_current_state(list_id) \
122			.all()
123
124			totals = {state: 0 for state in SubscriptionState}
125
126			for state_name, count in rows:
127			state = SubscriptionState[state_name]
128			totals[state] = count
129
130			totals['total'] = sum(totals.values())
131
132			return totals
133
134
135			def _build_query_for_current_state(list_id: ListID) -> BaseQuery:
136			"""Build a query to return the number of currently requested and
137			declined subscription states for that list.
138
139			The generated SQL should be equivalent to this:
140
141			SELECT
142			nso.state,
143			COUNT(nso.state)
144			FROM newsletter_subscriptions AS nso
145			JOIN (
146			SELECT
147			user_id,
148			list_id,
149			MAX(expressed_at) AS latest_expressed_at
150			FROM newsletter_subscriptions
151			GROUP BY
152			user_id,
153			list_id
154			) AS nsi
155			ON nso.user_id = nsi.user_id
156			AND nso.list_id = nsi.list_id
157			AND nso.expressed_at = nsi.latest_expressed_at
158			WHERE list_id = {list_id}
159			GROUP BY
160			list_id,
161			state
162			"""
163			subquery = _build_query_for_latest_expressed_at().subquery()
164
165			return db.session \
166			.query(
167			Subscription._state,
168			db.func.count(Subscription._state),
169			) \
170			.join(subquery, db.and_(
171			Subscription.user_id == subquery.c.user_id,
172			Subscription.list_id == subquery.c.list_id,
173			Subscription.expressed_at == subquery.c.latest_expressed_at
174			)) \
175			.filter_by(list_id=list_id) \
176			.group_by(
177			Subscription.list_id,
178			Subscription._state,
179			)
180
181
182			def _build_query_for_latest_expressed_at() -> BaseQuery:
183			"""Build a query to return the most recent time the subscription
184			state was set (grouped by user and list).
185
186			The generated SQL should be equivalent to this:
187
188			SELECT user_id, list_id, MAX(expressed_at) AS latest_expressed_at
189			FROM newsletter_subscriptions
190			GROUP BY user_id, list_id
191			"""
192			return db.session \
193			.query(
194			Subscription.user_id,
195			Subscription.list_id,
196			db.func.max(Subscription.expressed_at).label('latest_expressed_at')
197			) \
198			.group_by(
199			Subscription.user_id,
200			Subscription.list_id
201			)
202
203
204			def get_subscription_state(user_id: UserID, list_id: ListID
205			) -> SubscriptionState:
206			"""Return the user's current subscription state for that list."""
207			current_subscription = Subscription.query \
208			.filter_by(user_id=user_id) \
209			.filter_by(list_id=list_id) \
210			.order_by(Subscription.expressed_at.desc()) \
211			.first()
212
213			if current_subscription is None:
214			return SubscriptionState.declined
215
216			return current_subscription.state
217
218
219			def get_subscription_updates_for_user(user_id: UserID
220			) -> Sequence[Subscription]:
221			"""Return subscription updates made by the user, for any list."""
222			return Subscription.query \
223			.filter_by(user_id=user_id) \
224			.all()
225
226
227			def is_subscribed(user_id: UserID, list_id: ListID) -> bool:
228			"""Return if the user is subscribed to the list or not."""
229			subscription_state = get_subscription_state(user_id, list_id)
230			return subscription_state == SubscriptionState.requested
231
232
233			def _db_entity_to_list(list_: DbList) -> List:
234			return List(
235			list_.id,
236			list_.title,
237			)
238

byceps / byceps

Push — master ( 97b1b2...a0bd49 )

byceps.services.newsletter.service._db_entity_to_list() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like