byceps.services.user.command_service.initialize_account() - Code Metrics - Inspection of "Implement admin UI functionality to initialize a u..." - byceps/byceps - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( a23055...584f3d )

by Jochen

created 2019-09-23 18:01 UTC

initialize_account() A

↳ Parent: byceps.services.user.command_service

Complexity

Conditions

Size

Total Lines	18
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	9
nop	2
dl	0
loc	18
rs	9.95
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 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()


def enable_user(user_id: UserID, initiator_id: UserID) -> None:
    """Enable the user account."""
    user = _get_user(user_id)

    user.enabled = True

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

    db.session.commit()


def disable_user(user_id: UserID, initiator_id: UserID) -> None:
    """Disable the user account."""
    user = _get_user(user_id)

    user.enabled = False

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

    db.session.commit()


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

byceps / byceps

Push — master ( a23055...584f3d )

initialize_account() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like