Completed
Push — main ( 254773...79af2c )
by Jochen
03:19
created

create_user()   A

Complexity

Conditions 4

Size

Total Lines 45
Code Lines 31

Duplication

Lines 0
Ratio 0 %

Code Coverage

Tests 11
CRAP Score 4

Importance

Changes 0
Metric Value
cc 4
eloc 31
nop 9
dl 0
loc 45
ccs 11
cts 11
cp 1
crap 4
rs 9.1359
c 0
b 0
f 0

How to fix   Many Parameters   

Many Parameters

Methods with many parameters are not only hard to understand, but their parameters also often become inconsistent when you need more, or different data.

There are several approaches to avoid long parameter lists:

1
"""
2
byceps.services.user.creation_service
3
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
4
5
:Copyright: 2006-2020 Jochen Kupperschmidt
6
:License: Modified BSD, see LICENSE for details.
7
"""
8
9 1
from datetime import datetime
10 1
from typing import Optional, Set, Tuple
11
12 1
from flask import current_app
13
14 1
from ...database import db
15 1
from ...events.user import UserAccountCreated
16 1
from ...typing import UserID
17
18 1
from ..authentication.password import service as password_service
19 1
from ..consent import consent_service
20 1
from ..consent.transfer.models import Consent
21 1
from ..newsletter import command_service as newsletter_command_service
22 1
from ..newsletter.transfer.models import Subscription as NewsletterSubscription
23 1
from ..site.transfer.models import SiteID
24
25 1
from . import email_address_verification_service
26 1
from . import event_service
27 1
from .models.detail import UserDetail as DbUserDetail
28 1
from .models.user import User as DbUser
29 1
from . import service as user_service
30 1
from .transfer.models import User
31
32
33 1
class UserCreationFailed(Exception):
34 1
    pass
35
36
37 1
def create_user(
38
    screen_name: str,
39
    email_address: str,
40
    password: str,
41
    first_names: Optional[str],
42
    last_name: Optional[str],
43
    site_id: SiteID,
44
    *,
45
    consents: Set[Consent] = None,
46
    newsletter_subscription: Optional[NewsletterSubscription] = None,
47
) -> Tuple[User, UserAccountCreated]:
48
    """Create a user account and related records."""
49
    # user with details, password, and roles
50 1
    user, event = create_basic_user(
51
        screen_name,
52
        email_address,
53
        password,
54
        first_names=first_names,
55
        last_name=last_name,
56
    )
57
58
    # consents
59 1
    if consents:
60 1
        for consent in consents:
61
            # Insert missing user ID.
62 1
            consent = consent_service.build_consent(
63
                user.id,
64
                consent.subject_id,
65
                consent.expressed_at,
66
            )
67 1
            db.session.add(consent)
68
69 1
    db.session.commit()
70
71
    # newsletter subscription (optional)
72 1
    if newsletter_subscription:
73 1
        newsletter_command_service.subscribe(
74
            user.id,
75
            newsletter_subscription.list_id,
76
            newsletter_subscription.expressed_at,
77
        )
78
79 1
    request_email_address_confirmation(user, email_address, site_id)
80
81 1
    return user, event
82
83
84 1
def create_basic_user(
85
    screen_name: str,
86
    email_address: Optional[str],
87
    password: str,
88
    *,
89
    first_names: Optional[str] = None,
90
    last_name: Optional[str] = None,
91
    creator_id: Optional[UserID] = None,
92
) -> Tuple[User, UserAccountCreated]:
93
    # user with details
94 1
    user, event = _create_user(
95
        screen_name,
96
        email_address,
97
        first_names=first_names,
98
        last_name=last_name,
99
        creator_id=creator_id,
100
    )
101
102
    # password
103 1
    password_service.create_password_hash(user.id, password)
104
105 1
    return user, event
106
107
108 1
def _create_user(
109
    screen_name: Optional[str],
110
    email_address: Optional[str],
111
    *,
112
    first_names: Optional[str] = None,
113
    last_name: Optional[str] = None,
114
    creator_id: Optional[UserID] = None,
115
) -> Tuple[User, UserAccountCreated]:
116 1
    if creator_id is not None:
117
        creator = user_service.get_user(creator_id)
118
    else:
119 1
        creator = None
120
121 1
    created_at = datetime.utcnow()
122
123 1
    user = build_user(created_at, screen_name, email_address)
124
125 1
    user.detail.first_names = first_names
126 1
    user.detail.last_name = last_name
127
128 1
    db.session.add(user)
129
130 1
    try:
131 1
        db.session.commit()
132
    except Exception as e:
133
        current_app.logger.error('User creation failed: %s', e)
134
        db.session.rollback()
135
        raise UserCreationFailed()
136
137
    # Create event in separate step as user ID is not available earlier.
138 1
    event_data = {}
139 1
    if creator is not None:
140
        event_data['initiator_id'] = str(creator.id)
141 1
    event_service.create_event(
142
        'user-created', user.id, event_data, occurred_at=created_at
143
    )
144
145 1
    user_dto = user_service._db_entity_to_user(user)
146
147 1
    event = UserAccountCreated(
148
        occurred_at=user.created_at,
149
        initiator_id=creator.id if creator else None,
150
        initiator_screen_name=creator.screen_name if creator else None,
151
        user_id=user.id,
152
        user_screen_name=user.screen_name,
153
    )
154
155 1
    return user_dto, event
156
157
158 1
def build_user(
159
    created_at: datetime,
160
    screen_name: Optional[str],
161
    email_address: Optional[str],
162
) -> DbUser:
163 1
    if screen_name is not None:
164 1
        normalized_screen_name = _normalize_screen_name(screen_name)
165
    else:
166 1
        normalized_screen_name = None
167
168 1
    if email_address is not None:
169 1
        normalized_email_address = _normalize_email_address(email_address)
170
    else:
171
        normalized_email_address = None
172
173 1
    user = DbUser(created_at, normalized_screen_name, normalized_email_address)
174
175 1
    detail = DbUserDetail(user=user)
176
177 1
    return user
178
179
180 1
def request_email_address_confirmation(
181
    user: User, email_address: str, site_id: SiteID
182
) -> None:
183
    """Send an e-mail to the user to request confirmation of the e-mail
184
    address.
185
    """
186 1
    normalized_email_address = _normalize_email_address(email_address)
187
188 1
    email_address_verification_service.send_email_address_confirmation_email(
189
        normalized_email_address, user.screen_name, user.id, site_id
190
    )
191
192
193 1
def _normalize_screen_name(screen_name: str) -> str:
194
    """Normalize the screen name, or raise an exception if invalid."""
195 1
    normalized = screen_name.strip()
196
197 1
    if not normalized or (' ' in normalized) or ('@' in normalized):
198
        raise ValueError(f"Invalid screen name: '{screen_name}'")
199
200 1
    return normalized
201
202
203 1
def _normalize_email_address(email_address: str) -> str:
204
    """Normalize the e-mail address, or raise an exception if invalid."""
205 1
    normalized = email_address.strip().lower()
206
207 1
    if not normalized or (' ' in normalized) or ('@' not in normalized):
208
        raise ValueError(f"Invalid email address: '{email_address}'")
209
210
    return normalized
211