-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathuser.py
More file actions
169 lines (142 loc) · 4.96 KB
/
Copy pathuser.py
File metadata and controls
169 lines (142 loc) · 4.96 KB
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
from dataclasses import dataclass
from typing import Protocol
from src.core.security.password import PasswordSerrvice
from src.modules.authorization.domain.permissions import (
ADMIN_ROLE,
DEFAULT_USER_ROLE,
MANAGER_ROLE,
VIEWER_ROLE,
)
from src.modules.user.domain.entities.user import User, UserProfile
class SeedUserRepository(Protocol):
async def get_by_email(self, email: str) -> User | None:
raise NotImplementedError
async def save(self, user: User) -> User:
raise NotImplementedError
async def save_profile(self, profile: UserProfile) -> UserProfile:
raise NotImplementedError
class SeedAuthorizationService(Protocol):
async def assign_role(self, subject: str, role: str) -> None:
raise NotImplementedError
@dataclass(frozen=True)
class SeedUserConfig:
app_env: str
admin_email: str
admin_password: str
admin_username: str | None = None
admin_fullname: str | None = None
development_users_password: str = ""
@classmethod
def from_settings(cls, settings) -> "SeedUserConfig":
return cls(
app_env=settings.APP_ENV,
admin_email=settings.SEED_ADMIN_EMAIL,
admin_password=settings.SEED_ADMIN_PASSWORD,
admin_username=settings.SEED_ADMIN_USERNAME,
admin_fullname=settings.SEED_ADMIN_FULLNAME,
development_users_password=settings.SEED_DEVELOPMENT_USERS_PASSWORD,
)
@property
def has_admin_credentials(self) -> bool:
return bool(self.admin_email.strip() and self.admin_password.strip())
@property
def should_seed_development_users(self) -> bool:
return self.app_env.lower() == "development" and bool(
self.development_users_password.strip()
)
@dataclass(frozen=True)
class UserSeedResult:
users_created: int = 0
roles_assigned: int = 0
async def seed_user(
user_repository: SeedUserRepository,
authorization_service: SeedAuthorizationService,
config: SeedUserConfig,
) -> UserSeedResult:
users_created = 0
roles_assigned = 0
if not config.has_admin_credentials:
admin_result = UserSeedResult()
else:
admin_result = await _seed_one_user(
user_repository=user_repository,
authorization_service=authorization_service,
email=config.admin_email,
password=config.admin_password,
username=config.admin_username,
fullname=config.admin_fullname,
role=ADMIN_ROLE,
)
users_created += admin_result.users_created
roles_assigned += admin_result.roles_assigned
if config.should_seed_development_users:
for development_user in _development_users(config.development_users_password):
result = await _seed_one_user(
user_repository=user_repository,
authorization_service=authorization_service,
email=development_user.email,
password=development_user.password,
username=development_user.username,
fullname=development_user.fullname,
role=development_user.role,
)
users_created += result.users_created
roles_assigned += result.roles_assigned
return UserSeedResult(users_created=users_created, roles_assigned=roles_assigned)
@dataclass(frozen=True)
class DevelopmentSeedUser:
email: str
password: str
username: str
fullname: str
role: str
def _development_users(password: str) -> tuple[DevelopmentSeedUser, ...]:
return (
DevelopmentSeedUser(
email="user@example.com",
password=password,
username="user",
fullname="Default User",
role=DEFAULT_USER_ROLE,
),
DevelopmentSeedUser(
email="manager@example.com",
password=password,
username="manager",
fullname="Todo Manager",
role=MANAGER_ROLE,
),
DevelopmentSeedUser(
email="viewer@example.com",
password=password,
username="viewer",
fullname="Todo Viewer",
role=VIEWER_ROLE,
),
)
async def _seed_one_user(
user_repository: SeedUserRepository,
authorization_service: SeedAuthorizationService,
email: str,
password: str,
username: str | None,
fullname: str | None,
role: str,
) -> UserSeedResult:
existing = await user_repository.get_by_email(email)
if existing is not None:
return UserSeedResult()
user = User.create(
email=email,
password_hash=PasswordSerrvice.hash(password),
username=username,
)
saved_user = await user_repository.save(user)
await user_repository.save_profile(
UserProfile(
user_id=saved_user.id,
display_name=fullname,
)
)
await authorization_service.assign_role(str(saved_user.id), role)
return UserSeedResult(users_created=1, roles_assigned=1)