~homeworkprod/byceps

ref: 4237b3ec9496efe95dcce82bea3207ab9de4d520 byceps/byceps/services/user/event_service.py -rw-r--r-- 1.5 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
"""
byceps.services.user.event_service
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

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

from datetime import datetime
from typing import List, Optional

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

from .models.event import UserEvent, UserEventData


def create_event(
    event_type: str,
    user_id: UserID,
    data: UserEventData,
    *,
    occurred_at: Optional[datetime] = None,
) -> UserEvent:
    """Create a user event."""
    event = build_event(event_type, user_id, data, occurred_at=occurred_at)

    db.session.add(event)
    db.session.commit()

    return event


def build_event(
    event_type: str,
    user_id: UserID,
    data: UserEventData,
    *,
    occurred_at: Optional[datetime] = None,
) -> UserEvent:
    """Assemble, but not persist, a user event."""
    if occurred_at is None:
        occurred_at = datetime.utcnow()

    return UserEvent(occurred_at, event_type, user_id, data)


def get_events_for_user(user_id: UserID) -> List[UserEvent]:
    """Return the events for that user."""
    return UserEvent.query \
        .filter_by(user_id=user_id) \
        .order_by(UserEvent.occurred_at) \
        .all()


def get_events_of_type_for_user(
    event_type: str, user_id: UserID
) -> List[UserEvent]:
    """Return the events of that type for that user."""
    return UserEvent.query \
        .filter_by(user_id=user_id) \
        .filter_by(event_type=event_type) \
        .order_by(UserEvent.occurred_at) \
        .all()