Changed normal data collection to list comprehensions p3 #446

This commit is contained in:
Sven Heidemann 2023-12-02 18:16:04 +01:00
parent b6b9bfabf5
commit 0aa690b984
8 changed files with 152 additions and 321 deletions

View File

@ -48,66 +48,60 @@ class AchievementRepositoryService(AchievementRepositoryABC):
) )
def get_achievements(self) -> List[Achievement]: def get_achievements(self) -> List[Achievement]:
achievements = List(Achievement)
self._logger.trace(__name__, f"Send SQL command: {Achievement.get_select_all_string()}") self._logger.trace(__name__, f"Send SQL command: {Achievement.get_select_all_string()}")
results = self._context.select(Achievement.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Get user with id {result[0]}")
achievements.append(self._from_result(result))
return achievements return List(
Achievement,
[self._from_result(result) for result in self._context.select(Achievement.get_select_all_string())],
)
def get_achievement_by_id(self, id: int) -> Achievement: def get_achievement_by_id(self, id: int) -> Achievement:
self._logger.trace(__name__, f"Send SQL command: {Achievement.get_select_by_id_string(id)}") self._logger.trace(__name__, f"Send SQL command: {Achievement.get_select_by_id_string(id)}")
result = self._context.select(Achievement.get_select_by_id_string(id))[0] return self._from_result(self._context.select(Achievement.get_select_by_id_string(id))[0])
return self._from_result(result)
def get_achievements_by_server_id(self, server_id: int) -> List[Achievement]: def get_achievements_by_server_id(self, server_id: int) -> List[Achievement]:
achievements = List(Achievement)
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {Achievement.get_select_by_server_id_string(server_id)}", f"Send SQL command: {Achievement.get_select_by_server_id_string(server_id)}",
) )
results = self._context.select(Achievement.get_select_by_server_id_string(server_id))
for result in results:
self._logger.trace(__name__, f"Get user with id {result[0]}")
achievements.append(self._from_result(result))
return achievements return List(
Achievement,
[
self._from_result(result)
for result in self._context.select(Achievement.get_select_by_server_id_string(server_id))
],
)
def get_achievements_by_user_id(self, user_id: int) -> List[Achievement]: def get_achievements_by_user_id(self, user_id: int) -> List[Achievement]:
achievements = List(Achievement)
achievements_joins = List(UserGotAchievement)
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {UserGotAchievement.get_select_by_user_id_string(user_id)}", f"Send SQL command: {UserGotAchievement.get_select_by_user_id_string(user_id)}",
) )
results = self._context.select(UserGotAchievement.get_select_by_user_id_string(user_id))
for result in results:
self._logger.trace(__name__, f"Got UserGotAchievement with id {result[0]}")
achievements_joins.append(self._join_from_result(result))
for achievements_join in achievements_joins: return List(
results = self._context.select(Achievement.get_select_by_id_string(achievements_join.achievement.id)) UserGotAchievement,
for result in results: [
self._logger.trace(__name__, f"Got Achievement with id {result[0]}") self._join_from_result(result).achievement
achievements.append(self._from_result(result)) for result in self._context.select(UserGotAchievement.get_select_by_user_id_string(user_id))
],
return achievements )
def get_user_got_achievements_by_achievement_id(self, achievement_id: int) -> List[Achievement]: def get_user_got_achievements_by_achievement_id(self, achievement_id: int) -> List[Achievement]:
achievements_joins = List(UserGotAchievement)
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {UserGotAchievement.get_select_by_achievement_id_string(achievement_id)}", f"Send SQL command: {UserGotAchievement.get_select_by_achievement_id_string(achievement_id)}",
) )
results = self._context.select(UserGotAchievement.get_select_by_achievement_id_string(achievement_id))
for result in results:
self._logger.trace(__name__, f"Got UserGotAchievement with id {result[0]}")
achievements_joins.append(self._join_from_result(result))
return achievements_joins return List(
UserGotAchievement,
[
self._join_from_result(result)
for result in self._context.select(
UserGotAchievement.get_select_by_achievement_id_string(achievement_id)
)
],
)
def add_achievement(self, achievement: Achievement): def add_achievement(self, achievement: Achievement):
self._logger.trace(__name__, f"Send SQL command: {achievement.insert_string}") self._logger.trace(__name__, f"Send SQL command: {achievement.insert_string}")

View File

@ -44,13 +44,12 @@ class ApiKeyRepositoryService(ApiKeyRepositoryABC):
return api_key return api_key
def get_api_keys(self) -> List[ApiKey]: def get_api_keys(self) -> List[ApiKey]:
api_keys = List(ApiKey)
self._logger.trace(__name__, f"Send SQL command: {ApiKey.get_select_all_string()}") self._logger.trace(__name__, f"Send SQL command: {ApiKey.get_select_all_string()}")
results = self._context.select(ApiKey.get_select_all_string())
for result in results:
api_keys.append(self._api_key_from_result(result))
return api_keys return List(
ApiKey,
[self._api_key_from_result(result) for result in self._context.select(ApiKey.get_select_all_string())],
)
def get_api_key(self, identifier: str, key: str) -> ApiKey: def get_api_key(self, identifier: str, key: str) -> ApiKey:
self._logger.trace(__name__, f"Send SQL command: {ApiKey.get_select_string(identifier, key)}") self._logger.trace(__name__, f"Send SQL command: {ApiKey.get_select_string(identifier, key)}")

View File

@ -64,23 +64,16 @@ class AuthUserRepositoryService(AuthUserRepositoryABC):
__name__, __name__,
f"Send SQL command: {auth_user.get_select_user_id_from_relations()}", f"Send SQL command: {auth_user.get_select_user_id_from_relations()}",
) )
relation_ids = List(int)
results = self._context.select(auth_user.get_select_user_id_from_relations())
for result in results:
self._logger.trace(__name__, f"Got auth user relation with id {result[0]}")
relation_ids.append(result[0])
return relation_ids return List(int, [result[0] for result in self._context.select(auth_user.get_select_user_id_from_relations())])
def get_all_auth_users(self) -> List[AuthUser]: def get_all_auth_users(self) -> List[AuthUser]:
users = List(AuthUser)
self._logger.trace(__name__, f"Send SQL command: {AuthUser.get_select_all_string()}") self._logger.trace(__name__, f"Send SQL command: {AuthUser.get_select_all_string()}")
results = self._context.select(AuthUser.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Get auth user with id {result[0]}")
users.append(self._user_from_result(result))
return users return List(
AuthUser,
[self._user_from_result(result) for result in self._context.select(AuthUser.get_select_all_string())],
)
def get_filtered_auth_users(self, criteria: AuthUserSelectCriteria) -> FilteredResult: def get_filtered_auth_users(self, criteria: AuthUserSelectCriteria) -> FilteredResult:
users = self.get_all_auth_users() users = self.get_all_auth_users()
@ -129,8 +122,7 @@ class AuthUserRepositoryService(AuthUserRepositoryABC):
def get_auth_user_by_email(self, email: str) -> AuthUser: def get_auth_user_by_email(self, email: str) -> AuthUser:
self._logger.trace(__name__, f"Send SQL command: {AuthUser.get_select_by_email_string(email)}") self._logger.trace(__name__, f"Send SQL command: {AuthUser.get_select_by_email_string(email)}")
result = self._context.select(AuthUser.get_select_by_email_string(email))[0] return self._user_from_result(self._context.select(AuthUser.get_select_by_email_string(email))[0])
return self._user_from_result(result)
def find_auth_user_by_email(self, email: str) -> Optional[AuthUser]: def find_auth_user_by_email(self, email: str) -> Optional[AuthUser]:
self._logger.trace(__name__, f"Send SQL command: {AuthUser.get_select_by_email_string(email)}") self._logger.trace(__name__, f"Send SQL command: {AuthUser.get_select_by_email_string(email)}")
@ -138,9 +130,7 @@ class AuthUserRepositoryService(AuthUserRepositoryABC):
if result is None or len(result) == 0: if result is None or len(result) == 0:
return None return None
result = result[0] return self._user_from_result(result[0])
return self._user_from_result(result)
def find_auth_user_by_confirmation_id(self, id: str) -> Optional[AuthUser]: def find_auth_user_by_confirmation_id(self, id: str) -> Optional[AuthUser]:
self._logger.trace( self._logger.trace(
@ -151,9 +141,7 @@ class AuthUserRepositoryService(AuthUserRepositoryABC):
if result is None or len(result) == 0: if result is None or len(result) == 0:
return None return None
result = result[0] return self._user_from_result(result[0])
return self._user_from_result(result)
def find_auth_user_by_forgot_password_id(self, id: str) -> Optional[AuthUser]: def find_auth_user_by_forgot_password_id(self, id: str) -> Optional[AuthUser]:
self._logger.trace( self._logger.trace(
@ -164,9 +152,7 @@ class AuthUserRepositoryService(AuthUserRepositoryABC):
if result is None or len(result) == 0: if result is None or len(result) == 0:
return None return None
result = result[0] return self._user_from_result(result[0])
return self._user_from_result(result)
def add_auth_user(self, user: AuthUser): def add_auth_user(self, user: AuthUser):
self._logger.trace(__name__, f"Send SQL command: {user.insert_string}") self._logger.trace(__name__, f"Send SQL command: {user.insert_string}")

View File

@ -23,27 +23,7 @@ class AutoRoleRepositoryService(AutoRoleRepositoryABC):
AutoRoleRepositoryABC.__init__(self) AutoRoleRepositoryABC.__init__(self)
def get_auto_roles(self) -> List[AutoRole]: def _from_result(self, result: tuple) -> AutoRole:
auto_roles = List(AutoRole)
self._logger.trace(__name__, f"Send SQL command: {AutoRole.get_select_all_string()}")
results = self._context.select(AutoRole.get_select_all_string())
for result in results:
auto_roles.append(
AutoRole(
self._servers.get_server_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
)
return auto_roles
def get_auto_role_by_id(self, id: int) -> AutoRole:
self._logger.trace(__name__, f"Send SQL command: {AutoRole.get_select_by_id_string(id)}")
result = self._context.select(AutoRole.get_select_by_id_string(id))[0]
return AutoRole( return AutoRole(
self._servers.get_server_by_id(result[1]), self._servers.get_server_by_id(result[1]),
result[2], result[2],
@ -53,55 +33,37 @@ class AutoRoleRepositoryService(AutoRoleRepositoryABC):
id=result[0], id=result[0],
) )
def get_auto_roles(self) -> List[AutoRole]:
self._logger.trace(__name__, f"Send SQL command: {AutoRole.get_select_all_string()}")
return List(
AutoRole, [self._from_result(result) for result in self._context.select(AutoRole.get_select_all_string())]
)
def get_auto_role_by_id(self, id: int) -> AutoRole:
self._logger.trace(__name__, f"Send SQL command: {AutoRole.get_select_by_id_string(id)}")
return self._from_result(self._context.select(AutoRole.get_select_by_id_string(id))[0])
def find_auto_role_by_id(self, id: int) -> Optional[AutoRole]: def find_auto_role_by_id(self, id: int) -> Optional[AutoRole]:
self._logger.trace(__name__, f"Send SQL command: {AutoRole.get_select_by_id_string(id)}") self._logger.trace(__name__, f"Send SQL command: {AutoRole.get_select_by_id_string(id)}")
result = self._context.select(AutoRole.get_select_by_id_string(id)) result = self._context.select(AutoRole.get_select_by_id_string(id))
if result is None or len(result) == 0: if result is None or len(result) == 0:
return None return None
result = result[0] return self._from_result(result[0])
return AutoRole(
self._servers.get_server_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
def get_auto_roles_by_server_id(self, id: int) -> List[AutoRole]: def get_auto_roles_by_server_id(self, id: int) -> List[AutoRole]:
self._logger.trace(__name__, f"Send SQL command: {AutoRole.get_select_by_server_id_string(id)}") self._logger.trace(__name__, f"Send SQL command: {AutoRole.get_select_by_server_id_string(id)}")
auto_roles = List(AutoRole) return List(
results = self._context.select(AutoRole.get_select_by_server_id_string(id)) AutoRole,
for result in results: [self._from_result(result) for result in self._context.select(AutoRole.get_select_by_server_id_string(id))],
auto_roles.append(
AutoRole(
self._servers.get_server_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
) )
)
return auto_roles
def get_auto_role_by_message_id(self, id: int) -> AutoRole: def get_auto_role_by_message_id(self, id: int) -> AutoRole:
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {AutoRole.get_select_by_message_id_string(id)}", f"Send SQL command: {AutoRole.get_select_by_message_id_string(id)}",
) )
result = self._context.select(AutoRole.get_select_by_message_id_string(id))[0] return self._from_result(self._context.select(AutoRole.get_select_by_message_id_string(id))[0])
return AutoRole(
self._servers.get_server_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
def find_auto_role_by_message_id(self, id: int) -> Optional[AutoRole]: def find_auto_role_by_message_id(self, id: int) -> Optional[AutoRole]:
self._logger.trace( self._logger.trace(
@ -112,16 +74,7 @@ class AutoRoleRepositoryService(AutoRoleRepositoryABC):
if result is None or len(result) == 0: if result is None or len(result) == 0:
return None return None
result = result[0] return self._from_result(result[0])
return AutoRole(
self._servers.get_server_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
def add_auto_role(self, auto_role: AutoRole): def add_auto_role(self, auto_role: AutoRole):
self._logger.trace(__name__, f"Send SQL command: {auto_role.insert_string}") self._logger.trace(__name__, f"Send SQL command: {auto_role.insert_string}")
@ -135,27 +88,7 @@ class AutoRoleRepositoryService(AutoRoleRepositoryABC):
self._logger.trace(__name__, f"Send SQL command: {auto_role.delete_string}") self._logger.trace(__name__, f"Send SQL command: {auto_role.delete_string}")
self._context.cursor.execute(auto_role.delete_string) self._context.cursor.execute(auto_role.delete_string)
def get_auto_role_rules(self) -> List[AutoRoleRule]: def _rule_from_result(self, result: tuple) -> AutoRoleRule:
auto_role_rules = List(AutoRoleRule)
self._logger.trace(__name__, f"Send SQL command: {AutoRoleRule.get_select_all_string()}")
results = self._context.select(AutoRoleRule.get_select_all_string())
for result in results:
auto_role_rules.append(
AutoRoleRule(
self.get_auto_role_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
)
return auto_role_rules
def get_auto_role_rule_by_id(self, id: int) -> AutoRoleRule:
self._logger.trace(__name__, f"Send SQL command: {AutoRoleRule.get_select_by_id_string(id)}")
result = self._context.select(AutoRoleRule.get_select_by_id_string(id))[0]
return AutoRoleRule( return AutoRoleRule(
self.get_auto_role_by_id(result[1]), self.get_auto_role_by_id(result[1]),
result[2], result[2],
@ -165,26 +98,29 @@ class AutoRoleRepositoryService(AutoRoleRepositoryABC):
id=result[0], id=result[0],
) )
def get_auto_role_rules(self) -> List[AutoRoleRule]:
self._logger.trace(__name__, f"Send SQL command: {AutoRoleRule.get_select_all_string()}")
return List(
AutoRoleRule,
[self._rule_from_result(result) for result in self._context.select(AutoRoleRule.get_select_all_string())],
)
def get_auto_role_rule_by_id(self, id: int) -> AutoRoleRule:
self._logger.trace(__name__, f"Send SQL command: {AutoRoleRule.get_select_by_id_string(id)}")
return self._rule_from_result(self._context.select(AutoRoleRule.get_select_by_id_string(id))[0])
def get_auto_role_rules_by_auto_role_id(self, id: int) -> List[AutoRoleRule]: def get_auto_role_rules_by_auto_role_id(self, id: int) -> List[AutoRoleRule]:
auto_role_rules = List(AutoRoleRule)
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {AutoRoleRule.get_select_by_auto_role_id_string(id)}", f"Send SQL command: {AutoRoleRule.get_select_by_auto_role_id_string(id)}",
) )
results = self._context.select(AutoRoleRule.get_select_by_auto_role_id_string(id)) return List(
for result in results: AutoRoleRule,
auto_role_rules.append( [
AutoRoleRule( self._rule_from_result(result)
self.get_auto_role_by_id(result[1]), for result in self._context.select(AutoRoleRule.get_select_by_auto_role_id_string(id))
result[2], ],
result[3],
result[4],
result[5],
id=result[0],
) )
)
return auto_role_rules
def add_auto_role_rule(self, auto_role_rule: AutoRoleRule): def add_auto_role_rule(self, auto_role_rule: AutoRoleRule):
self._logger.trace(__name__, f"Send SQL command: {auto_role_rule.insert_string}") self._logger.trace(__name__, f"Send SQL command: {auto_role_rule.insert_string}")

View File

@ -23,32 +23,7 @@ class ClientRepositoryService(ClientRepositoryABC):
ClientRepositoryABC.__init__(self) ClientRepositoryABC.__init__(self)
def get_clients(self) -> List[Client]: def _from_result(self, result: tuple) -> Client:
clients = List(Client)
self._logger.trace(__name__, f"Send SQL command: {Client.get_select_all_string()}")
results = self._context.select(Client.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Get client with id {result[0]}")
clients.append(
Client(
result[1],
result[2],
result[3],
result[4],
result[5],
result[6],
self._servers.get_server_by_id(result[7]),
result[8],
result[9],
id=result[0],
)
)
return clients
def get_client_by_id(self, client_id: int) -> Client:
self._logger.trace(__name__, f"Send SQL command: {Client.get_select_by_id_string(client_id)}")
result = self._context.select(Client.get_select_by_id_string(client_id))
return Client( return Client(
result[1], result[1],
result[2], result[2],
@ -62,49 +37,37 @@ class ClientRepositoryService(ClientRepositoryABC):
id=result[0], id=result[0],
) )
def get_clients(self) -> List[Client]:
self._logger.trace(__name__, f"Send SQL command: {Client.get_select_all_string()}")
return List(
Client, [self._from_result(result) for result in self._context.select(Client.get_select_all_string())]
)
def get_client_by_id(self, client_id: int) -> Client:
self._logger.trace(__name__, f"Send SQL command: {Client.get_select_by_id_string(client_id)}")
return self._from_result(self._context.select(Client.get_select_by_id_string(client_id))[0])
def get_clients_by_server_id(self, server_id: int) -> List[Client]: def get_clients_by_server_id(self, server_id: int) -> List[Client]:
clients = List(Client)
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {Client.get_select_by_server_id_string(server_id)}", f"Send SQL command: {Client.get_select_by_server_id_string(server_id)}",
) )
results = self._context.select(Client.get_select_by_server_id_string(server_id))
for result in results:
clients.append(
Client(
result[1],
result[2],
result[3],
result[4],
result[5],
result[6],
self._servers.get_server_by_id(result[7]),
result[8],
result[9],
id=result[0],
)
)
return clients return List(
Client,
[
self._from_result(result)
for result in self._context.select(Client.get_select_by_server_id_string(server_id))
],
)
def get_client_by_discord_id(self, discord_id: int) -> Client: def get_client_by_discord_id(self, discord_id: int) -> Client:
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {Client.get_select_by_discord_id_string(discord_id)}", f"Send SQL command: {Client.get_select_by_discord_id_string(discord_id)}",
) )
result = self._context.select(Client.get_select_by_discord_id_string(discord_id))[0] return self._from_result(self._context.select(Client.get_select_by_discord_id_string(discord_id))[0])
return Client(
result[1],
result[2],
result[3],
result[4],
result[5],
result[6],
self._servers.get_server_by_id(result[7]),
result[8],
result[9],
id=result[0],
)
def find_client_by_discord_id(self, discord_id: int) -> Optional[Client]: def find_client_by_discord_id(self, discord_id: int) -> Optional[Client]:
self._logger.trace( self._logger.trace(
@ -115,20 +78,7 @@ class ClientRepositoryService(ClientRepositoryABC):
if result is None or len(result) == 0: if result is None or len(result) == 0:
return None return None
result = result[0] return self._from_result(result[0])
return Client(
result[1],
result[2],
result[3],
result[4],
result[5],
result[6],
self._servers.get_server_by_id(result[7]),
result[8],
result[9],
id=result[0],
)
def find_client_by_server_id(self, discord_id: int) -> Optional[Client]: def find_client_by_server_id(self, discord_id: int) -> Optional[Client]:
self._logger.trace( self._logger.trace(
@ -139,20 +89,7 @@ class ClientRepositoryService(ClientRepositoryABC):
if result is None or len(result) == 0: if result is None or len(result) == 0:
return None return None
result = result[0] return self._from_result(result[0])
return Client(
result[1],
result[2],
result[3],
result[4],
result[5],
result[6],
self._servers.get_server_by_id(result[7]),
result[8],
result[9],
id=result[0],
)
def find_client_by_discord_id_and_server_id(self, discord_id: int, server_id: int) -> Optional[Client]: def find_client_by_discord_id_and_server_id(self, discord_id: int, server_id: int) -> Optional[Client]:
self._logger.trace( self._logger.trace(
@ -163,20 +100,7 @@ class ClientRepositoryService(ClientRepositoryABC):
if result is None or len(result) == 0: if result is None or len(result) == 0:
return None return None
result = result[0] return self._from_result(result[0])
return Client(
result[1],
result[2],
result[3],
result[4],
result[5],
result[6],
self._servers.get_server_by_id(result[7]),
result[8],
result[9],
id=result[0],
)
def add_client(self, client: Client): def add_client(self, client: Client):
self._logger.trace(__name__, f"Send SQL command: {client.insert_string}") self._logger.trace(__name__, f"Send SQL command: {client.insert_string}")

View File

@ -35,49 +35,49 @@ class GameServerRepositoryService(GameServerRepositoryABC):
) )
def get_game_servers(self) -> List[GameServer]: def get_game_servers(self) -> List[GameServer]:
game_servers = List(GameServer)
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {GameServer.get_select_all_string()}", f"Send SQL command: {GameServer.get_select_all_string()}",
) )
results = self._context.select(GameServer.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Get user-joined-game-server with id {result[0]}")
game_servers.append(self._from_result(result))
return game_servers return List(
GameServer,
[self._from_result(result) for result in self._context.select(GameServer.get_select_all_string())],
)
def get_game_servers_by_server_id(self, id: int) -> List[GameServer]: def get_game_servers_by_server_id(self, id: int) -> List[GameServer]:
game_servers = List(GameServer)
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {GameServer.get_select_by_server_id_string(id)}", f"Send SQL command: {GameServer.get_select_by_server_id_string(id)}",
) )
results = self._context.select(GameServer.get_select_by_server_id_string(id))
for result in results:
self._logger.trace(__name__, f"Get user-joined-game-server with id {result[0]}")
game_servers.append(self._from_result(result))
return game_servers return List(
GameServer,
[
self._from_result(result)
for result in self._context.select(GameServer.get_select_by_server_id_string(id))
],
)
def get_game_server_by_id(self, id: int) -> GameServer: def get_game_server_by_id(self, id: int) -> GameServer:
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {GameServer.get_select_by_id_string(id)}", f"Send SQL command: {GameServer.get_select_by_id_string(id)}",
) )
result = self._context.select(GameServer.get_select_by_id_string(id))[0] return self._from_result(self._context.select(GameServer.get_select_by_id_string(id))[0])
return self._from_result(result)
def get_game_servers_by_api_key_id(self, id: int) -> List[GameServer]: def get_game_servers_by_api_key_id(self, id: int) -> List[GameServer]:
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {GameServer.get_select_by_api_key_id_string(id)}", f"Send SQL command: {GameServer.get_select_by_api_key_id_string(id)}",
) )
game_servers = List(GameServer) return List(
results = self._context.select(GameServer.get_select_by_api_key_id_string(id)) GameServer,
for result in results: [
game_servers.append(self._from_result(result)) self._from_result(result)
return game_servers for result in self._context.select(GameServer.get_select_by_api_key_id_string(id))
],
)
def add_game_server(self, game_server: GameServer): def add_game_server(self, game_server: GameServer):
self._logger.trace(__name__, f"Send SQL command: {game_server.insert_string}") self._logger.trace(__name__, f"Send SQL command: {game_server.insert_string}")

View File

@ -4,8 +4,8 @@ from cpl_core.database.context import DatabaseContextABC
from cpl_query.extension import List from cpl_query.extension import List
from bot_core.logging.database_logger import DatabaseLogger from bot_core.logging.database_logger import DatabaseLogger
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.abc.level_repository_abc import LevelRepositoryABC from bot_data.abc.level_repository_abc import LevelRepositoryABC
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.model.level import Level from bot_data.model.level import Level
@ -41,20 +41,15 @@ class LevelRepositoryService(LevelRepositoryABC):
) )
def get_levels(self) -> List[Level]: def get_levels(self) -> List[Level]:
levels = List(Level)
self._logger.trace(__name__, f"Send SQL command: {Level.get_select_all_string()}") self._logger.trace(__name__, f"Send SQL command: {Level.get_select_all_string()}")
results = self._context.select(Level.get_select_all_string()) return List(
for result in results: Level, [self._level_from_result(result) for result in self._context.select(Level.get_select_all_string())]
self._logger.trace(__name__, f"Get level with id {result[0]}") )
levels.append(self._level_from_result(result))
return levels
def get_level_by_id(self, id: int) -> Level: def get_level_by_id(self, id: int) -> Level:
self._logger.trace(__name__, f"Send SQL command: {Level.get_select_by_id_string(id)}") self._logger.trace(__name__, f"Send SQL command: {Level.get_select_by_id_string(id)}")
result = self._context.select(Level.get_select_by_id_string(id))[0]
return self._level_from_result(result) return self._level_from_result(self._context.select(Level.get_select_by_id_string(id))[0])
def find_level_by_id(self, id: int) -> Optional[Level]: def find_level_by_id(self, id: int) -> Optional[Level]:
self._logger.trace(__name__, f"Send SQL command: {Level.get_select_by_id_string(id)}") self._logger.trace(__name__, f"Send SQL command: {Level.get_select_by_id_string(id)}")
@ -65,21 +60,19 @@ class LevelRepositoryService(LevelRepositoryABC):
return self._level_from_result(result[0]) return self._level_from_result(result[0])
def get_levels_by_server_id(self, server_id: int) -> List[Level]: def get_levels_by_server_id(self, server_id: int) -> List[Level]:
levels = List(Level)
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {Level.get_select_by_server_id_string(server_id)}", f"Send SQL command: {Level.get_select_by_server_id_string(server_id)}",
) )
results = self._context.select(Level.get_select_by_server_id_string(server_id)) return List(
Level,
for result in results: [
self._logger.trace(__name__, f"Get level with id {result[0]}") self._level_from_result(result)
levels.append(self._level_from_result(result)) for result in self._context.select(Level.get_select_by_server_id_string(server_id))
],
return levels )
def find_levels_by_server_id(self, server_id: int) -> Optional[List[Level]]: def find_levels_by_server_id(self, server_id: int) -> Optional[List[Level]]:
levels = List(Level)
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {Level.get_select_by_server_id_string(server_id)}", f"Send SQL command: {Level.get_select_by_server_id_string(server_id)}",
@ -88,11 +81,10 @@ class LevelRepositoryService(LevelRepositoryABC):
if results is None or len(results) == 0: if results is None or len(results) == 0:
return None return None
for result in results: return List(
self._logger.trace(__name__, f"Get level with id {result[0]}") Level,
levels.append(self._level_from_result(result)) [self._level_from_result(result) for result in results],
)
return levels
def add_level(self, level: Level): def add_level(self, level: Level):
self._logger.trace(__name__, f"Send SQL command: {level.insert_string}") self._logger.trace(__name__, f"Send SQL command: {level.insert_string}")

View File

@ -5,8 +5,8 @@ from cpl_query.extension import List
from discord import EntityType from discord import EntityType
from bot_core.logging.database_logger import DatabaseLogger from bot_core.logging.database_logger import DatabaseLogger
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.abc.scheduled_event_repository_abc import ScheduledEventRepositoryABC from bot_data.abc.scheduled_event_repository_abc import ScheduledEventRepositoryABC
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.model.scheduled_event import ScheduledEvent from bot_data.model.scheduled_event import ScheduledEvent
from bot_data.model.scheduled_event_interval_enum import ScheduledEventIntervalEnum from bot_data.model.scheduled_event_interval_enum import ScheduledEventIntervalEnum
@ -49,34 +49,34 @@ class ScheduledEventRepositoryService(ScheduledEventRepositoryABC):
) )
def get_scheduled_events(self) -> List[ScheduledEvent]: def get_scheduled_events(self) -> List[ScheduledEvent]:
scheduled_events = List(ScheduledEvent)
self._logger.trace(__name__, f"Send SQL command: {ScheduledEvent.get_select_all_string()}") self._logger.trace(__name__, f"Send SQL command: {ScheduledEvent.get_select_all_string()}")
results = self._context.select(ScheduledEvent.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Get scheduled_event with id {result[0]}")
scheduled_events.append(self._scheduled_event_from_result(result))
return scheduled_events return List(
ScheduledEvent,
[
self._scheduled_event_from_result(result)
for result in self._context.select(ScheduledEvent.get_select_all_string())
],
)
def get_scheduled_event_by_id(self, id: int) -> ScheduledEvent: def get_scheduled_event_by_id(self, id: int) -> ScheduledEvent:
self._logger.trace(__name__, f"Send SQL command: {ScheduledEvent.get_select_by_id_string(id)}") self._logger.trace(__name__, f"Send SQL command: {ScheduledEvent.get_select_by_id_string(id)}")
result = self._context.select(ScheduledEvent.get_select_by_id_string(id))[0]
return self._scheduled_event_from_result(result) return self._scheduled_event_from_result(self._context.select(ScheduledEvent.get_select_by_id_string(id))[0])
def get_scheduled_events_by_server_id(self, server_id: int) -> List[ScheduledEvent]: def get_scheduled_events_by_server_id(self, server_id: int) -> List[ScheduledEvent]:
scheduled_events = List(ScheduledEvent)
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {ScheduledEvent.get_select_by_server_id_string(server_id)}", f"Send SQL command: {ScheduledEvent.get_select_by_server_id_string(server_id)}",
) )
results = self._context.select(ScheduledEvent.get_select_by_server_id_string(server_id))
for result in results: return List(
self._logger.trace(__name__, f"Get scheduled_event with id {result[0]}") ScheduledEvent,
scheduled_events.append(self._scheduled_event_from_result(result)) [
self._scheduled_event_from_result(result)
return scheduled_events for result in self._context.select(ScheduledEvent.get_select_by_server_id_string(server_id))
],
)
def add_scheduled_event(self, scheduled_event: ScheduledEvent): def add_scheduled_event(self, scheduled_event: ScheduledEvent):
self._logger.trace(__name__, f"Send SQL command: {scheduled_event.insert_string}") self._logger.trace(__name__, f"Send SQL command: {scheduled_event.insert_string}")