byceps.services.user.command_service._assign_roles() - Code Metrics - Inspection of "Assign initial user roles on account initializatio..." - byceps/byceps - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 259657...0d00f7 )

by Jochen

created 2019-10-06 12:54 UTC

_assign_roles() A

↳ Parent: byceps.services.user.command_service

Complexity

Conditions

Size

Total Lines	5
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

"""
byceps.services.user.command_service
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

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

from datetime import date
from typing import Optional

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

from ..authorization.models import RoleID
from ..authorization import service as authorization_service

from . import event_service
from .models.detail import UserDetail as DbUserDetail
from .models.user import User as DbUser


def initialize_account(user_id: UserID, initiator_id: UserID) -> None:
    """Initialize the user account.

    This is meant to happen only once at most, and can not be undone.
    """
    user = _get_user(user_id)

    if user.initialized:
        raise ValueError(f'Account is already initialized.')

    user.initialized = True

    event = event_service.build_event('user-initialized', user.id, {
        'initiator_id': str(initiator_id),
    })
    db.session.add(event)

    db.session.commit()

    _assign_roles(user.id, initiator_id)


def _assign_roles(user_id: UserID, initiator_id: UserID) -> None:
    board_user_role = authorization_service.find_role(RoleID('board_user'))

    authorization_service.assign_role_to_user(
        board_user_role.id, user_id, initiator_id=initiator_id
    )


def suspend_account(user_id: UserID, initiator_id: UserID, reason: str) -> None:
    """Suspend the user account."""
    user = _get_user(user_id)

    user.suspended = True

    event = event_service.build_event('user-suspended', user.id, {
        'initiator_id': str(initiator_id),
        'reason': reason,
    })
    db.session.add(event)

    db.session.commit()


def unsuspend_account(
    user_id: UserID, initiator_id: UserID, reason: str
) -> None:
    """Unsuspend the user account."""
    user = _get_user(user_id)

    user.suspended = False

    event = event_service.build_event('user-unsuspended', user.id, {
        'initiator_id': str(initiator_id),
        'reason': reason,
    })
    db.session.add(event)

    db.session.commit()


def delete_account(user_id: UserID, initiator_id: UserID, reason: str) -> None:
    """Delete the user account."""
    user = _get_user(user_id)

    user.deleted = True
    _anonymize_account(user)

    event = event_service.build_event('user-deleted', user.id, {
        'initiator_id': str(initiator_id),
        'reason': reason,
    })
    db.session.add(event)

    # Deassign authorization roles.
    authorization_service.deassign_all_roles_from_user(
        user.id, initiator_id, commit=False
    )

    db.session.commit()


def change_screen_name(
    user_id: UserID,
    new_screen_name: str,
    initiator_id: UserID,
    *,
    reason: Optional[str] = None,
) -> None:
    """Change the user's screen name."""
    user = _get_user(user_id)

    old_screen_name = user.screen_name

    user.screen_name = new_screen_name

    event_data = {
        'old_screen_name': old_screen_name,
        'new_screen_name': new_screen_name,
        'initiator_id': str(initiator_id),
    }
    if reason:
        event_data['reason'] = reason

    event = event_service.build_event(
        'user-screen-name-changed', user.id, event_data
    )
    db.session.add(event)

    db.session.commit()


def change_email_address(
    user_id: UserID,
    new_email_address: str,
    initiator_id: UserID,
    *,
    reason: Optional[str] = None,
) -> None:
    """Change the user's e-mail address."""
    user = _get_user(user_id)

    old_email_address = user.email_address

    user.email_address = new_email_address
    user.email_address_verified = False

    event_data = {
        'old_email_address': old_email_address,
        'new_email_address': new_email_address,
        'initiator_id': str(initiator_id),
    }
    if reason:
        event_data['reason'] = reason

    event = event_service.build_event(
        'user-email-address-changed', user.id, event_data
    )
    db.session.add(event)

    db.session.commit()


def update_user_details(
    user_id: UserID,
    first_names: str,
    last_name: str,
    date_of_birth: date,
    country: str,
    zip_code,
    city: str,
    street: str,
    phone_number: str,
) -> None:
    """Update the user's details."""
    detail = _get_user_detail(user_id)

    detail.first_names = first_names
    detail.last_name = last_name
    detail.date_of_birth = date_of_birth
    detail.country = country
    detail.zip_code = zip_code
    detail.city = city
    detail.street = street
    detail.phone_number = phone_number

    db.session.commit()


def set_user_detail_extra(user_id: UserID, key: str, value: str) -> None:
    """Set a value for a key in the user's detail extras map."""
    detail = _get_user_detail(user_id)

    if detail.extras is None:
        detail.extras = {}

    detail.extras[key] = value

    db.session.commit()


def remove_user_detail_extra(user_id: UserID, key: str) -> None:
    """Remove the entry with that key from the user's detail extras map."""
    detail = _get_user_detail(user_id)

    if (detail.extras is None) or (key not in detail.extras):
        return

    del detail.extras[key]
    db.session.commit()


def _anonymize_account(user: DbUser) -> None:
    """Remove or replace user details of the account."""
    user.screen_name = f'deleted-{user.id.hex}'
    user.email_address = f'{user.id.hex}@user.invalid'
    user.legacy_id = None

    # Remove details.
    user.detail.first_names = None
    user.detail.last_name = None
    user.detail.date_of_birth = None
    user.detail.country = None
    user.detail.zip_code = None
    user.detail.city = None
    user.detail.street = None
    user.detail.phone_number = None

    # Remove avatar association.
    if user.avatar_selection is not None:
        db.session.delete(user.avatar_selection)


def _get_user(user_id: UserID) -> DbUser:
    """Return the user with that ID, or raise an exception."""
    user = DbUser.query.get(user_id)

    if user is None:
        raise ValueError(f"Unknown user ID '{user_id}'")

    return user


def _get_user_detail(user_id: UserID) -> DbUserDetail:
    """Return the user's details, or raise an exception."""
    detail = DbUserDetail.query \
        .filter_by(user_id=user_id) \
        .one_or_none()

    if detail is None:
        raise ValueError(f"Unknown user ID '{user_id}'")

    return detail


1			"""
2			byceps.services.user.command_service
3			~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
4
5			:Copyright: 2006-2019 Jochen Kupperschmidt
6			:License: Modified BSD, see LICENSE for details.
7			"""
8
9			from datetime import date
10			from typing import Optional
11
12			from ...database import db
13			from ...typing import UserID
14
15			from ..authorization.models import RoleID
16			from ..authorization import service as authorization_service
17
18			from . import event_service
19			from .models.detail import UserDetail as DbUserDetail
20			from .models.user import User as DbUser
21
22
23			def initialize_account(user_id: UserID, initiator_id: UserID) -> None:
24			"""Initialize the user account.
25
26			This is meant to happen only once at most, and can not be undone.
27			"""
28			user = _get_user(user_id)
29
30			if user.initialized:
31			raise ValueError(f'Account is already initialized.')
32
33			user.initialized = True
34
35			event = event_service.build_event('user-initialized', user.id, {
36			'initiator_id': str(initiator_id),
37			})
38			db.session.add(event)
39
40			db.session.commit()
41
42			_assign_roles(user.id, initiator_id)
43
44
45			def _assign_roles(user_id: UserID, initiator_id: UserID) -> None:
46			board_user_role = authorization_service.find_role(RoleID('board_user'))
47
48			authorization_service.assign_role_to_user(
49			board_user_role.id, user_id, initiator_id=initiator_id
50			)
51
52
53			def suspend_account(user_id: UserID, initiator_id: UserID, reason: str) -> None:
54			"""Suspend the user account."""
55			user = _get_user(user_id)
56
57			user.suspended = True
58
59			event = event_service.build_event('user-suspended', user.id, {
60			'initiator_id': str(initiator_id),
61			'reason': reason,
62			})
63			db.session.add(event)
64
65			db.session.commit()
66
67
68			def unsuspend_account(
69			user_id: UserID, initiator_id: UserID, reason: str
70			) -> None:
71			"""Unsuspend the user account."""
72			user = _get_user(user_id)
73
74			user.suspended = False
75
76			event = event_service.build_event('user-unsuspended', user.id, {
77			'initiator_id': str(initiator_id),
78			'reason': reason,
79			})
80			db.session.add(event)
81
82			db.session.commit()
83
84
85			def delete_account(user_id: UserID, initiator_id: UserID, reason: str) -> None:
86			"""Delete the user account."""
87			user = _get_user(user_id)
88
89			user.deleted = True
90			_anonymize_account(user)
91
92			event = event_service.build_event('user-deleted', user.id, {
93			'initiator_id': str(initiator_id),
94			'reason': reason,
95			})
96			db.session.add(event)
97
98			# Deassign authorization roles.
99			authorization_service.deassign_all_roles_from_user(
100			user.id, initiator_id, commit=False
101			)
102
103			db.session.commit()
104
105
106			def change_screen_name(
107			user_id: UserID,
108			new_screen_name: str,
109			initiator_id: UserID,
110			*,
111			reason: Optional[str] = None,
112			) -> None:
113			"""Change the user's screen name."""
114			user = _get_user(user_id)
115
116			old_screen_name = user.screen_name
117
118			user.screen_name = new_screen_name
119
120			event_data = {
121			'old_screen_name': old_screen_name,
122			'new_screen_name': new_screen_name,
123			'initiator_id': str(initiator_id),
124			}
125			if reason:
126			event_data['reason'] = reason
127
128			event = event_service.build_event(
129			'user-screen-name-changed', user.id, event_data
130			)
131			db.session.add(event)
132
133			db.session.commit()
134
135
136			def change_email_address(
137			user_id: UserID,
138			new_email_address: str,
139			initiator_id: UserID,
140			*,
141			reason: Optional[str] = None,
142			) -> None:
143			"""Change the user's e-mail address."""
144			user = _get_user(user_id)
145
146			old_email_address = user.email_address
147
148			user.email_address = new_email_address
149			user.email_address_verified = False
150
151			event_data = {
152			'old_email_address': old_email_address,
153			'new_email_address': new_email_address,
154			'initiator_id': str(initiator_id),
155			}
156			if reason:
157			event_data['reason'] = reason
158
159			event = event_service.build_event(
160			'user-email-address-changed', user.id, event_data
161			)
162			db.session.add(event)
163
164			db.session.commit()
165
166
167			def update_user_details(
168			user_id: UserID,
169			first_names: str,
170			last_name: str,
171			date_of_birth: date,
172			country: str,
173			zip_code,
174			city: str,
175			street: str,
176			phone_number: str,
177			) -> None:
178			"""Update the user's details."""
179			detail = _get_user_detail(user_id)
180
181			detail.first_names = first_names
182			detail.last_name = last_name
183			detail.date_of_birth = date_of_birth
184			detail.country = country
185			detail.zip_code = zip_code
186			detail.city = city
187			detail.street = street
188			detail.phone_number = phone_number
189
190			db.session.commit()
191
192
193			def set_user_detail_extra(user_id: UserID, key: str, value: str) -> None:
194			"""Set a value for a key in the user's detail extras map."""
195			detail = _get_user_detail(user_id)
196
197			if detail.extras is None:
198			detail.extras = {}
199
200			detail.extras[key] = value
201
202			db.session.commit()
203
204
205			def remove_user_detail_extra(user_id: UserID, key: str) -> None:
206			"""Remove the entry with that key from the user's detail extras map."""
207			detail = _get_user_detail(user_id)
208
209			if (detail.extras is None) or (key not in detail.extras):
210			return
211
212			del detail.extras[key]
213			db.session.commit()
214
215
216			def _anonymize_account(user: DbUser) -> None:
217			"""Remove or replace user details of the account."""
218			user.screen_name = f'deleted-{user.id.hex}'
219			user.email_address = f'{user.id.hex}@user.invalid'
220			user.legacy_id = None
221
222			# Remove details.
223			user.detail.first_names = None
224			user.detail.last_name = None
225			user.detail.date_of_birth = None
226			user.detail.country = None
227			user.detail.zip_code = None
228			user.detail.city = None
229			user.detail.street = None
230			user.detail.phone_number = None
231
232			# Remove avatar association.
233			if user.avatar_selection is not None:
234			db.session.delete(user.avatar_selection)
235
236
237			def _get_user(user_id: UserID) -> DbUser:
238			"""Return the user with that ID, or raise an exception."""
239			user = DbUser.query.get(user_id)
240
241			if user is None:
242			raise ValueError(f"Unknown user ID '{user_id}'")
243
244			return user
245
246
247			def _get_user_detail(user_id: UserID) -> DbUserDetail:
248			"""Return the user's details, or raise an exception."""
249			detail = DbUserDetail.query \
250			.filter_by(user_id=user_id) \
251			.one_or_none()
252
253			if detail is None:
254			raise ValueError(f"Unknown user ID '{user_id}'")
255
256			return detail
257

byceps / byceps

Push — master ( 259657...0d00f7 )

_assign_roles() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like