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