~homeworkprod/byceps

ref: d5f15dd3d7905ef733a0c3201f8a82cf18c1fe22 byceps/scripts/clean_up_after_deleted_users.py -rwxr-xr-x 5.9 KiB
d5f15dd3 — Jochen Kupperschmidt Fix import of `RecentLogin` in script 8 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
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
#!/usr/bin/env python

"""Remove data for user accounts that have been marked as deleted.

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

from typing import Callable, Set

import click

from byceps.database import db
from byceps.services.authentication.password.models import Credential
from byceps.services.authentication.session.models.recent_login import (
    RecentLogin,
)
from byceps.services.authentication.session.models.session_token import (
    SessionToken,
)
from byceps.services.authorization.models import UserRole
from byceps.services.board.models.last_category_view import (
    LastCategoryView as BoardLastCategoryView,
)
from byceps.services.board.models.last_topic_view import (
    LastTopicView as BoardLastTopicView,
)
from byceps.services.consent.models.consent import Consent
from byceps.services.newsletter.models import (
    SubscriptionUpdate as NewsletterSubscriptionUpdate,
)
from byceps.services.user.models.event import UserEvent
from byceps.services.user import service as user_service
from byceps.services.user_avatar.models import (
    AvatarSelection as UserAvatarSelection,
)
from byceps.services.verification_token.models import Token as VerificationToken
from byceps.typing import UserID
from byceps.util.system import get_config_filename_from_env_or_exit

from _util import app_context


@click.command()
@click.option(
    '--dry-run',
    is_flag=True,
    help='determine but do not delete affected records',
)
@click.argument('user_ids', nargs=-1, required=True)
def execute(dry_run, user_ids):
    user_ids = set(user_ids)

    check_for_undeleted_accounts(user_ids)

    def delete(label: str, delete_func: Callable) -> None:
        delete_records(label, delete_func, user_ids)

    delete('authentication credentials', delete_authn_credentials)
    delete('recent logins', delete_authn_recent_logins)
    delete('session tokens', delete_authn_session_tokens)
    delete('authorization role assignments', delete_authz_user_roles)
    delete('board category view marks', delete_board_category_lastviews)
    delete('board topic view marks', delete_board_topic_lastviews)
    delete('consents', delete_consents)
    delete(
        'newsletter subscription updates',
        delete_newsletter_subscription_updates,
    )
    delete('user avatar selections', delete_user_avatar_selections)
    delete('user events', delete_user_events)
    delete('verification tokens', delete_verification_tokens)

    if not dry_run:
        db.session.commit()


def check_for_undeleted_accounts(user_ids: Set[UserID]) -> None:
    users = user_service.find_users(user_ids)

    non_deleted_users = [u for u in users if not u.deleted]
    if non_deleted_users:
        user_ids_string = ', '.join(str(u.id) for u in non_deleted_users)
        raise click.BadParameter(
            f'These user accounts are not marked as deleted: {user_ids_string}'
        )


def delete_records(
    label: str, delete_func: Callable, user_ids: Set[UserID]
) -> None:
    click.secho(f'Deleting {label} ... ', nl=False)
    affected = delete_func(user_ids)
    click.secho(str(affected), fg='yellow')


def delete_authn_credentials(user_ids: Set[UserID]) -> int:
    """Delete authentication credentials for the given users."""
    return _execute_delete_for_users_query(Credential, user_ids)


def delete_authn_recent_logins(user_ids: Set[UserID]) -> int:
    """Delete recent logins for the given users."""
    return _execute_delete_for_users_query(RecentLogin, user_ids)


def delete_authn_session_tokens(user_ids: Set[UserID]) -> int:
    """Delete session tokens for the given users."""
    return _execute_delete_for_users_query(SessionToken, user_ids)


def delete_authz_user_roles(user_ids: Set[UserID]) -> int:
    """Delete authorization role assignments from the given users."""
    return _execute_delete_for_users_query(UserRole, user_ids)


def delete_board_category_lastviews(user_ids: Set[UserID]) -> int:
    """Delete last board category view marks for the given users."""
    return _execute_delete_for_users_query(BoardLastCategoryView, user_ids)


def delete_board_topic_lastviews(user_ids: Set[UserID]) -> int:
    """Delete last board topic view marks for the given users."""
    return _execute_delete_for_users_query(BoardLastTopicView, user_ids)


def delete_consents(user_ids: Set[UserID]) -> int:
    """Delete consents from the given users."""
    return _execute_delete_for_users_query(Consent, user_ids)


def delete_newsletter_subscription_updates(user_ids: Set[UserID]) -> int:
    """Delete newsletter subscription updates for the given users."""
    return _execute_delete_for_users_query(
        NewsletterSubscriptionUpdate, user_ids
    )


def delete_user_avatar_selections(user_ids: Set[UserID]) -> int:
    """Delete user avatar selections (but not user avatar records and
    image files at this point) for the given users.
    """
    return _execute_delete_for_users_query(UserAvatarSelection, user_ids)


def delete_user_events(user_ids: Set[UserID]) -> int:
    """Delete user events (execpt for those that justify the deletion)
    for the given users.
    """
    return (
        db.session.query(UserEvent)
        .filter(UserEvent.user_id.in_(user_ids))
        .filter(UserEvent.event_type != 'user-deleted')
        .delete(synchronize_session=False)
    )


def delete_verification_tokens(user_ids: Set[UserID]) -> int:
    """Delete verification tokens for the given users."""
    return _execute_delete_for_users_query(VerificationToken, user_ids)


def _execute_delete_for_users_query(model, user_ids: Set[UserID]) -> int:
    """Execute (but not commit) deletions, return number of affected rows."""
    return (
        db.session.query(model)
        .filter(model.user_id.in_(user_ids))
        .delete(synchronize_session=False)
    )


if __name__ == '__main__':
    config_filename = get_config_filename_from_env_or_exit()
    with app_context(config_filename):
        execute()