clean_up_after_deleted_users.delete_newsletter_subscriptions() - Code Metrics - Inspection of "Recompile translations" - byceps/byceps - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — main ( 419964...c34879 )

by Jochen

created 2022-04-20 23:38 UTC

delete_newsletter_subscriptions() A

↳ Parent: clean_up_after_deleted_users

Complexity

Conditions

Size

Total Lines	3
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

#!/usr/bin/env python

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

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

from __future__ import annotations
from typing import Callable

import click

from byceps.database import db
from byceps.services.authentication.password.dbmodels import Credential
from byceps.services.authentication.session.dbmodels.recent_login import (
    RecentLogin,
)
from byceps.services.authentication.session.dbmodels.session_token import (
    SessionToken,
)
from byceps.services.authorization.dbmodels import UserRole
from byceps.services.board.dbmodels.last_category_view import (
    LastCategoryView as BoardLastCategoryView,
)
from byceps.services.board.dbmodels.last_topic_view import (
    LastTopicView as BoardLastTopicView,
)
from byceps.services.consent.dbmodels.consent import Consent
from byceps.services.newsletter.dbmodels import (
    Subscription as NewsletterSubscription,
    SubscriptionUpdate as NewsletterSubscriptionUpdate,
)
from byceps.services.user.dbmodels.log import UserLogEntry as DbUserLogEntry
from byceps.services.user import service as user_service
from byceps.services.user_avatar.dbmodels import (
    AvatarSelection as UserAvatarSelection,
)
from byceps.services.verification_token.dbmodels import (
    Token as VerificationToken,
)
from byceps.typing import UserID

from _util import call_with_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) -> None:
    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 subscriptions', delete_newsletter_subscriptions)
    delete(
        'newsletter subscription updates',
        delete_newsletter_subscription_updates,
    )
    delete('user avatar selections', delete_user_avatar_selections)
    delete('user log entries', delete_user_log_entries)
    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.get_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_subscriptions(user_ids: set[UserID]) -> int:
    """Delete newsletter subscriptions for the given users."""
    return _execute_delete_for_users_query(NewsletterSubscription, 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_log_entries(user_ids: set[UserID]) -> int:
    """Delete user log entries (except for those that justify the
    deletion) for the given users.
    """
    return (
        db.session.query(DbUserLogEntry)
        .filter(DbUserLogEntry.user_id.in_(user_ids))
        .filter(DbUserLogEntry.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__':
    call_with_app_context(execute)


1			#!/usr/bin/env python
2
3			"""Remove data for user accounts that have been marked as deleted.
4
5			:Copyright: 2014-2022 Jochen Kupperschmidt
6			:License: Revised BSD (see `LICENSE` file for details)
7			"""
8
9			from __future__ import annotations
10			from typing import Callable
11
12			import click
13
14			from byceps.database import db
15			from byceps.services.authentication.password.dbmodels import Credential
16			from byceps.services.authentication.session.dbmodels.recent_login import (
17			RecentLogin,
18			)
19			from byceps.services.authentication.session.dbmodels.session_token import (
20			SessionToken,
21			)
22			from byceps.services.authorization.dbmodels import UserRole
23			from byceps.services.board.dbmodels.last_category_view import (
24			LastCategoryView as BoardLastCategoryView,
25			)
26			from byceps.services.board.dbmodels.last_topic_view import (
27			LastTopicView as BoardLastTopicView,
28			)
29			from byceps.services.consent.dbmodels.consent import Consent
30			from byceps.services.newsletter.dbmodels import (
31			Subscription as NewsletterSubscription,
32			SubscriptionUpdate as NewsletterSubscriptionUpdate,
33			)
34			from byceps.services.user.dbmodels.log import UserLogEntry as DbUserLogEntry
35			from byceps.services.user import service as user_service
36			from byceps.services.user_avatar.dbmodels import (
37			AvatarSelection as UserAvatarSelection,
38			)
39			from byceps.services.verification_token.dbmodels import (
40			Token as VerificationToken,
41			)
42			from byceps.typing import UserID
43
44			from _util import call_with_app_context
45
46
47			@click.command()
48			@click.option(
49			'--dry-run',
50			is_flag=True,
51			help='determine but do not delete affected records',
52			)
53			@click.argument('user_ids', nargs=-1, required=True)
54			def execute(dry_run, user_ids) -> None:
55			user_ids = set(user_ids)
56
57			check_for_undeleted_accounts(user_ids)
58
59			def delete(label: str, delete_func: Callable) -> None:
60			delete_records(label, delete_func, user_ids)
61
62			delete('authentication credentials', delete_authn_credentials)
63			delete('recent logins', delete_authn_recent_logins)
64			delete('session tokens', delete_authn_session_tokens)
65			delete('authorization role assignments', delete_authz_user_roles)
66			delete('board category view marks', delete_board_category_lastviews)
67			delete('board topic view marks', delete_board_topic_lastviews)
68			delete('consents', delete_consents)
69			delete('newsletter subscriptions', delete_newsletter_subscriptions)
70			delete(
71			'newsletter subscription updates',
72			delete_newsletter_subscription_updates,
73			)
74			delete('user avatar selections', delete_user_avatar_selections)
75			delete('user log entries', delete_user_log_entries)
76			delete('verification tokens', delete_verification_tokens)
77
78			if not dry_run:
79			db.session.commit()
80
81
82			def check_for_undeleted_accounts(user_ids: set[UserID]) -> None:
83			users = user_service.get_users(user_ids)
84
85			non_deleted_users = [u for u in users if not u.deleted]
86			if non_deleted_users:
87			user_ids_string = ', '.join(str(u.id) for u in non_deleted_users)
88			raise click.BadParameter(
89			f'These user accounts are not marked as deleted: {user_ids_string}'
90			)
91
92
93			def delete_records(
94			label: str, delete_func: Callable, user_ids: set[UserID]
95			) -> None:
96			click.secho(f'Deleting {label} ... ', nl=False)
97			affected = delete_func(user_ids)
98			click.secho(str(affected), fg='yellow')
99
100
101			def delete_authn_credentials(user_ids: set[UserID]) -> int:
102			"""Delete authentication credentials for the given users."""
103			return _execute_delete_for_users_query(Credential, user_ids)
104
105
106			def delete_authn_recent_logins(user_ids: set[UserID]) -> int:
107			"""Delete recent logins for the given users."""
108			return _execute_delete_for_users_query(RecentLogin, user_ids)
109
110
111			def delete_authn_session_tokens(user_ids: set[UserID]) -> int:
112			"""Delete session tokens for the given users."""
113			return _execute_delete_for_users_query(SessionToken, user_ids)
114
115
116			def delete_authz_user_roles(user_ids: set[UserID]) -> int:
117			"""Delete authorization role assignments from the given users."""
118			return _execute_delete_for_users_query(UserRole, user_ids)
119
120
121			def delete_board_category_lastviews(user_ids: set[UserID]) -> int:
122			"""Delete last board category view marks for the given users."""
123			return _execute_delete_for_users_query(BoardLastCategoryView, user_ids)
124
125
126			def delete_board_topic_lastviews(user_ids: set[UserID]) -> int:
127			"""Delete last board topic view marks for the given users."""
128			return _execute_delete_for_users_query(BoardLastTopicView, user_ids)
129
130
131			def delete_consents(user_ids: set[UserID]) -> int:
132			"""Delete consents from the given users."""
133			return _execute_delete_for_users_query(Consent, user_ids)
134
135
136			def delete_newsletter_subscriptions(user_ids: set[UserID]) -> int:
137			"""Delete newsletter subscriptions for the given users."""
138			return _execute_delete_for_users_query(NewsletterSubscription, user_ids)
139
140
141			def delete_newsletter_subscription_updates(user_ids: set[UserID]) -> int:
142			"""Delete newsletter subscription updates for the given users."""
143			return _execute_delete_for_users_query(
144			NewsletterSubscriptionUpdate, user_ids
145			)
146
147
148			def delete_user_avatar_selections(user_ids: set[UserID]) -> int:
149			"""Delete user avatar selections (but not user avatar records and
150			image files at this point) for the given users.
151			"""
152			return _execute_delete_for_users_query(UserAvatarSelection, user_ids)
153
154
155			def delete_user_log_entries(user_ids: set[UserID]) -> int:
156			"""Delete user log entries (except for those that justify the
157			deletion) for the given users.
158			"""
159			return (
160			db.session.query(DbUserLogEntry)
161			.filter(DbUserLogEntry.user_id.in_(user_ids))
162			.filter(DbUserLogEntry.event_type != 'user-deleted')
163			.delete(synchronize_session=False)
164			)
165
166
167			def delete_verification_tokens(user_ids: set[UserID]) -> int:
168			"""Delete verification tokens for the given users."""
169			return _execute_delete_for_users_query(VerificationToken, user_ids)
170
171
172			def _execute_delete_for_users_query(model, user_ids: set[UserID]) -> int:
173			"""Execute (but not commit) deletions, return number of affected rows."""
174			return (
175			db.session.query(model)
176			.filter(model.user_id.in_(user_ids))
177			.delete(synchronize_session=False)
178			)
179
180
181			if __name__ == '__main__':
182			call_with_app_context(execute)
183

byceps / byceps

Push — main ( 419964...c34879 )

delete_newsletter_subscriptions() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like