~homeworkprod/byceps

ref: 4237b3ec9496efe95dcce82bea3207ab9de4d520 byceps/byceps/services/newsletter/command_service.py -rw-r--r-- 2.1 KiB
4237b3ec — Jochen Kupperschmidt Move ticketing blueprint into `site` subpackage 1 year, 11 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
"""
byceps.services.newsletter.command_service
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

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

from datetime import datetime

from ...database import db
from ...typing import UserID

from .models import List as DbList, SubscriptionUpdate as DbSubscriptionUpdate
from .service import find_list, _db_entity_to_list
from .transfer.models import List, ListID
from .types import SubscriptionState


class UnknownListId(Exception):
    pass


def create_list(list_id: ListID, title: str) -> List:
    """Create a list."""
    list_ = DbList(list_id, title)

    db.session.add(list_)
    db.session.commit()

    return _db_entity_to_list(list_)


def delete_list(list_id: ListID) -> None:
    """Delete a list."""
    db.session.query(DbList) \
        .filter_by(id=list_id) \
        .delete()

    db.session.commit()


def subscribe(user_id: UserID, list_id: ListID, expressed_at: datetime) -> None:
    """Subscribe the user to that list."""
    _update_subscription_state(
        user_id, list_id, expressed_at, SubscriptionState.requested
    )


def unsubscribe(
    user_id: UserID, list_id: ListID, expressed_at: datetime
) -> None:
    """Unsubscribe the user from that list."""
    _update_subscription_state(
        user_id, list_id, expressed_at, SubscriptionState.declined
    )


def _update_subscription_state(
    user_id: UserID,
    list_id: ListID,
    expressed_at: datetime,
    state: SubscriptionState,
) -> None:
    """Update the user's subscription state for that list."""
    list_ = find_list(list_id)
    if list_ is None:
        raise UnknownListId(list_id)

    subscription_update = DbSubscriptionUpdate(
        user_id, list_.id, expressed_at, state
    )

    db.session.add(subscription_update)
    db.session.commit()


def delete_subscription_updates(user_id: UserID, list_id: ListID) -> None:
    """Delete all subscription updates for the user and list."""
    db.session.query(DbSubscriptionUpdate) \
        .filter_by(user_id=user_id, list_id=list_id) \
        .delete()

    db.session.commit()