118 lines
3.5 KiB
Python
118 lines
3.5 KiB
Python
from backend.code.enums.errors_enum import ErrorTypeEnum
|
|
from backend.code.modules.database.database_sqlite import DatabaseSqliteModule
|
|
from backend.code.modules.users.repository import UserRepository
|
|
from backend.code.modules.accounts.repository import AccountRepository
|
|
from backend.code.models.responses.user_detail import UserDetailResponse
|
|
|
|
class UserController():
|
|
def __init__(self):
|
|
database = DatabaseSqliteModule()
|
|
self.repo = UserRepository(database)
|
|
self.account_repo = AccountRepository(database)
|
|
|
|
def ListUsers(self, params: str = None):
|
|
result = []
|
|
|
|
# Process filters
|
|
if params is not None:
|
|
params = self._OnProcessParams(params)
|
|
|
|
users = self.repo.FilterUsers(params)
|
|
if len(users) == 0:
|
|
return result
|
|
|
|
for _user in users:
|
|
user = UserDetailResponse(
|
|
Id = _user[0],
|
|
Name = _user[1],
|
|
Surnames = _user[2],
|
|
Email = _user[3],
|
|
Username = _user[4],
|
|
Locked = _user[5],
|
|
Active = _user[6]
|
|
)
|
|
result.append(user)
|
|
|
|
|
|
return result
|
|
|
|
def SelectUser(self, id: int):
|
|
user = self.repo.SelectUserById(id)
|
|
if user is None:
|
|
return
|
|
|
|
return UserDetailResponse(
|
|
Id = user[0],
|
|
Name = user[1],
|
|
Surnames = user[2],
|
|
Email = user[3],
|
|
Username = user[4],
|
|
Locked = user[5],
|
|
Active = user[6]
|
|
)
|
|
|
|
def RegisterUser(self, name: str, surnames: str, email: str, username: str = None, locked: bool = False, active: bool = True):
|
|
create_account = (username is not None)
|
|
|
|
# Check not exists
|
|
user_exists = self.repo.FilterUsers({'email': email})
|
|
if len(user_exists) > 0:
|
|
return ErrorTypeEnum.USER_EMAIL_EXISTS
|
|
|
|
if create_account:
|
|
account_exists = self.account_repo.SelectByUsername(username)
|
|
if account_exists is not None:
|
|
return ErrorTypeEnum.USER_ACCOUNT_EXISTS
|
|
|
|
# Register
|
|
user_id = self.repo.InsertUser(name, surnames, email)
|
|
if user_id is None:
|
|
return
|
|
|
|
if create_account:
|
|
account_id = self.account_repo.InsertAccount(user_id, username, locked, active)
|
|
if account_id is None:
|
|
return
|
|
|
|
return UserDetailResponse(
|
|
Id = user_id,
|
|
Name = name,
|
|
Surnames = surnames,
|
|
Email = email,
|
|
Username = username,
|
|
Locked = locked,
|
|
Active = active
|
|
)
|
|
|
|
def EditUser(self, id: int, name: str, surnames: str, email: str):
|
|
exists = self.repo.SelectUserById(id)
|
|
if exists is None:
|
|
return ErrorTypeEnum.USER_ID_NOT_EXISTS
|
|
|
|
edited = self.repo.UpdateUser(id, name, surnames, email)
|
|
if edited == 0:
|
|
return
|
|
|
|
return True
|
|
|
|
def DeleteUser(self, id: int):
|
|
exists = self.repo.SelectUserById(id)
|
|
if exists is None:
|
|
return ErrorTypeEnum.USER_ID_NOT_EXISTS
|
|
|
|
deleted = self.repo.DeleteUser(id)
|
|
if deleted == 0:
|
|
return
|
|
|
|
return True
|
|
|
|
#region Utilities
|
|
def _OnProcessParams(self, params: str):
|
|
result = {}
|
|
params = params.strip()
|
|
params = params.split(',')
|
|
for param in params:
|
|
field,value=param.split('=')
|
|
result[field.lower()]=value
|
|
return result
|
|
#endregion |