Changed normal data collection to list comprehensions p2 #446

This commit is contained in:
Sven Heidemann 2023-12-02 17:53:23 +01:00
parent d927ab8fb7
commit b6b9bfabf5
6 changed files with 86 additions and 99 deletions

View File

@ -26,15 +26,14 @@ class ServerConfigRepositoryService(ServerConfigRepositoryABC):
self._servers = servers self._servers = servers
def _get_team_role_ids(self, server_id: int) -> List[ServerTeamRoleIdsConfig]: def _get_team_role_ids(self, server_id: int) -> List[ServerTeamRoleIdsConfig]:
ids = List(ServerTeamRoleIdsConfig)
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {ServerTeamRoleIdsConfig.get_select_by_server_id_string(server_id)}", f"Send SQL command: {ServerTeamRoleIdsConfig.get_select_by_server_id_string(server_id)}",
) )
results = self._context.select(ServerTeamRoleIdsConfig.get_select_by_server_id_string(server_id))
for result in results: return List(
self._logger.trace(__name__, f"Got ServerTeamRoleIdsConfig with id {result[0]}") ServerTeamRoleIdsConfig,
ids.append( [
ServerTeamRoleIdsConfig( ServerTeamRoleIdsConfig(
result[1], result[1],
TeamMemberTypeEnum(result[2]), TeamMemberTypeEnum(result[2]),
@ -43,22 +42,23 @@ class ServerConfigRepositoryService(ServerConfigRepositoryABC):
result[5], result[5],
id=result[0], id=result[0],
) )
) for result in self._context.select(ServerTeamRoleIdsConfig.get_select_by_server_id_string(server_id))
],
return ids )
def _get_afk_channel_ids(self, server_id: int) -> List[int]: def _get_afk_channel_ids(self, server_id: int) -> List[int]:
urls = List(int)
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {ServerAFKChannelIdsConfig.get_select_by_server_id_string(server_id)}", f"Send SQL command: {ServerAFKChannelIdsConfig.get_select_by_server_id_string(server_id)}",
) )
results = self._context.select(ServerAFKChannelIdsConfig.get_select_by_server_id_string(server_id))
for result in results:
self._logger.trace(__name__, f"Got ServerAFKChannelIdsConfig with id {result[0]}")
urls.append(result[1])
return urls return List(
int,
[
result[1]
for result in self._context.select(ServerAFKChannelIdsConfig.get_select_by_server_id_string(server_id))
],
)
def _from_result(self, result: tuple) -> ServerConfig: def _from_result(self, result: tuple) -> ServerConfig:
return ServerConfig( return ServerConfig(
@ -102,18 +102,14 @@ class ServerConfigRepositoryService(ServerConfigRepositoryABC):
__name__, __name__,
f"Send SQL command: {ServerConfig.get_select_by_server_id_string(server_id)}", f"Send SQL command: {ServerConfig.get_select_by_server_id_string(server_id)}",
) )
result = self._context.select(ServerConfig.get_select_by_server_id_string(server_id))[0] return self._from_result(self._context.select(ServerConfig.get_select_by_server_id_string(server_id))[0])
return self._from_result(result)
def get_server_config_by_id(self, config_id: int) -> ServerConfig: def get_server_config_by_id(self, config_id: int) -> ServerConfig:
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {ServerConfig.get_select_by_id_string(config_id)}", f"Send SQL command: {ServerConfig.get_select_by_id_string(config_id)}",
) )
result = self._context.select(ServerConfig.get_select_by_id_string(config_id))[0] return self._from_result(self._context.select(ServerConfig.get_select_by_id_string(config_id))[0])
return self._from_result(result)
def add_server_config(self, server_config: ServerConfig): def add_server_config(self, server_config: ServerConfig):
self._logger.trace(__name__, f"Send SQL command: {server_config.insert_string}") self._logger.trace(__name__, f"Send SQL command: {server_config.insert_string}")

View File

@ -25,12 +25,15 @@ class ServerRepositoryService(ServerRepositoryABC):
ServerRepositoryABC.__init__(self) ServerRepositoryABC.__init__(self)
def get_servers(self) -> List[Server]: def get_servers(self) -> List[Server]:
servers = List(Server)
self._logger.trace(__name__, f"Send SQL command: {Server.get_select_all_string()}") self._logger.trace(__name__, f"Send SQL command: {Server.get_select_all_string()}")
results = self._context.select(Server.get_select_all_string())
for result in results:
servers.append(Server(result[1], result[2], result[3], id=result[0]))
servers = List(
Server,
[
Server(result[1], result[2], result[3], id=result[0])
for result in self._context.select(Server.get_select_all_string())
],
)
self._cache.add_servers(servers) self._cache.add_servers(servers)
return servers return servers
@ -96,9 +99,7 @@ class ServerRepositoryService(ServerRepositoryABC):
if result is None or len(result) == 0: if result is None or len(result) == 0:
return None return None
result = result[0] return Server(result[0][1], result[0][2], result[0][3], id=result[0][0])
return Server(result[1], result[2], result[3], id=result[0])
def add_server(self, server: Server): def add_server(self, server: Server):
self._logger.trace(__name__, f"Send SQL command: {server.insert_string}") self._logger.trace(__name__, f"Send SQL command: {server.insert_string}")

View File

@ -43,14 +43,15 @@ class ShortRoleNameRepositoryService(ShortRoleNameRepositoryABC):
) )
def get_short_role_names(self) -> List[ShortRoleName]: def get_short_role_names(self) -> List[ShortRoleName]:
short_role_names = List(ShortRoleName)
self._logger.trace(__name__, f"Send SQL command: {ShortRoleName.get_select_all_string()}") self._logger.trace(__name__, f"Send SQL command: {ShortRoleName.get_select_all_string()}")
results = self._context.select(ShortRoleName.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Get short_role_name with id {result[0]}")
short_role_names.append(self._short_role_name_from_result(result))
return short_role_names return List(
ShortRoleName,
[
self._short_role_name_from_result(result)
for result in self._context.select(ShortRoleName.get_select_all_string())
],
)
def get_short_role_name_by_id(self, id: int) -> ShortRoleName: def get_short_role_name_by_id(self, id: int) -> ShortRoleName:
self._logger.trace(__name__, f"Send SQL command: {ShortRoleName.get_select_by_id_string(id)}") self._logger.trace(__name__, f"Send SQL command: {ShortRoleName.get_select_by_id_string(id)}")
@ -59,31 +60,30 @@ class ShortRoleNameRepositoryService(ShortRoleNameRepositoryABC):
return self._short_role_name_from_result(result) return self._short_role_name_from_result(result)
def find_short_role_names_by_role_id(self, role_id: int) -> List[ShortRoleName]: def find_short_role_names_by_role_id(self, role_id: int) -> List[ShortRoleName]:
short_role_names = List(ShortRoleName)
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {ShortRoleName.get_select_by_role_id_string(role_id)}", f"Send SQL command: {ShortRoleName.get_select_by_role_id_string(role_id)}",
) )
results = self._context.select(ShortRoleName.get_select_by_role_id_string(role_id)) return List(
for result in results: ShortRoleName,
self._logger.trace(__name__, f"Get short_role_name with id {result[0]}") [
short_role_names.append(self._short_role_name_from_result(result)) self._short_role_name_from_result(result)
for result in self._context.select(ShortRoleName.get_select_by_role_id_string(role_id))
return short_role_names ],
)
def get_short_role_names_by_server_id(self, server_id: int) -> List[ShortRoleName]: def get_short_role_names_by_server_id(self, server_id: int) -> List[ShortRoleName]:
short_role_names = List(ShortRoleName)
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {ShortRoleName.get_select_by_server_id_string(server_id)}", f"Send SQL command: {ShortRoleName.get_select_by_server_id_string(server_id)}",
) )
results = self._context.select(ShortRoleName.get_select_by_server_id_string(server_id)) return List(
ShortRoleName,
for result in results: [
self._logger.trace(__name__, f"Get short_role_name with id {result[0]}") self._short_role_name_from_result(result)
short_role_names.append(self._short_role_name_from_result(result)) for result in self._context.select(ShortRoleName.get_select_by_server_id_string(server_id))
],
return short_role_names )
def add_short_role_name(self, short_role_name: ShortRoleName): def add_short_role_name(self, short_role_name: ShortRoleName):
self._logger.trace(__name__, f"Send SQL command: {short_role_name.insert_string}") self._logger.trace(__name__, f"Send SQL command: {short_role_name.insert_string}")

View File

@ -42,23 +42,24 @@ class SteamSpecialOfferRepositoryService(SteamSpecialOfferRepositoryABC):
) )
def get_steam_special_offers(self) -> List[SteamSpecialOffer]: def get_steam_special_offers(self) -> List[SteamSpecialOffer]:
steam_special_offers = List(SteamSpecialOffer)
self._logger.trace(__name__, f"Send SQL command: {SteamSpecialOffer.get_select_all_string()}") self._logger.trace(__name__, f"Send SQL command: {SteamSpecialOffer.get_select_all_string()}")
results = self._context.select(SteamSpecialOffer.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Get steam_special_offer with id {result[0]}")
steam_special_offers.append(self._steam_special_offer_from_result(result))
return steam_special_offers return List(
SteamSpecialOffer,
[
self._steam_special_offer_from_result(result)
for result in self._context.select(SteamSpecialOffer.get_select_all_string())
],
)
def get_steam_special_offer_by_name(self, name: str) -> SteamSpecialOffer: def get_steam_special_offer_by_name(self, name: str) -> SteamSpecialOffer:
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {SteamSpecialOffer.get_select_by_name_string(name)}", f"Send SQL command: {SteamSpecialOffer.get_select_by_name_string(name)}",
) )
result = self._context.select(SteamSpecialOffer.get_select_by_name_string(name))[0] return self._steam_special_offer_from_result(
self._context.select(SteamSpecialOffer.get_select_by_name_string(name))[0]
return self._steam_special_offer_from_result(result) )
def add_steam_special_offer(self, steam_special_offer: SteamSpecialOffer): def add_steam_special_offer(self, steam_special_offer: SteamSpecialOffer):
self._logger.trace(__name__, f"Send SQL command: {steam_special_offer.insert_string}") self._logger.trace(__name__, f"Send SQL command: {steam_special_offer.insert_string}")

View File

@ -18,27 +18,17 @@ class TechnicianConfigRepositoryService(TechnicianConfigRepositoryABC):
self._context = db_context self._context = db_context
def _get_technician_ids(self) -> List[int]: def _get_technician_ids(self) -> List[int]:
ids = List(int)
self._logger.trace(__name__, f"Send SQL command: {TechnicianIdConfig.get_select_all_string()}") self._logger.trace(__name__, f"Send SQL command: {TechnicianIdConfig.get_select_all_string()}")
results = self._context.select(TechnicianIdConfig.get_select_all_string()) return List(int, [config[1] for config in self._context.select(TechnicianIdConfig.get_select_all_string())])
for result in results:
self._logger.trace(__name__, f"Got TechnicianId with id {result[0]}")
ids.append(result[1])
return ids
def _get_technician_ping_urls(self) -> List[str]: def _get_technician_ping_urls(self) -> List[str]:
urls = List(str)
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {TechnicianPingUrlConfig.get_select_all_string()}", f"Send SQL command: {TechnicianPingUrlConfig.get_select_all_string()}",
) )
results = self._context.select(TechnicianPingUrlConfig.get_select_all_string()) return List(
for result in results: str, [ping_url[1] for ping_url in self._context.select(TechnicianPingUrlConfig.get_select_all_string())]
self._logger.trace(__name__, f"Got TechnicianPingUrl with id {result[0]}") )
urls.append(result[1])
return urls
def _from_result(self, result: tuple) -> TechnicianConfig: def _from_result(self, result: tuple) -> TechnicianConfig:
return TechnicianConfig( return TechnicianConfig(
@ -64,9 +54,7 @@ class TechnicianConfigRepositoryService(TechnicianConfigRepositoryABC):
def get_technician_config(self) -> TechnicianConfig: def get_technician_config(self) -> TechnicianConfig:
self._logger.trace(__name__, f"Send SQL command: {TechnicianConfig.get_select_all_string()}") self._logger.trace(__name__, f"Send SQL command: {TechnicianConfig.get_select_all_string()}")
result = self._context.select(TechnicianConfig.get_select_all_string())[0] return self._from_result(self._context.select(TechnicianConfig.get_select_all_string())[0])
return self._from_result(result)
def add_technician_config(self, technician_config: TechnicianConfig): def add_technician_config(self, technician_config: TechnicianConfig):
self._logger.trace(__name__, f"Send SQL command: {technician_config.insert_string}") self._logger.trace(__name__, f"Send SQL command: {technician_config.insert_string}")

View File

@ -39,46 +39,47 @@ class UserGameIdentRepositoryService(UserGameIdentRepositoryABC):
) )
def get_user_game_idents(self) -> List[UserGameIdent]: def get_user_game_idents(self) -> List[UserGameIdent]:
joins = List(UserGameIdent)
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {UserGameIdent.get_select_all_string()}", f"Send SQL command: {UserGameIdent.get_select_all_string()}",
) )
results = self._context.select(UserGameIdent.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Get UserGameIdent with id {result[0]}")
joins.append(self._from_result(result))
return joins return List(
UserGameIdent,
[
self._from_result(game_ident)
for game_ident in self._context.select(UserGameIdent.get_select_all_string())
],
)
def get_user_game_idents_by_game_server_id(self, game_server_id: int) -> List[UserGameIdent]: def get_user_game_idents_by_game_server_id(self, game_server_id: int) -> List[UserGameIdent]:
joins = List(UserGameIdent)
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {UserGameIdent.get_select_by_game_server_id_string(game_server_id)}", f"Send SQL command: {UserGameIdent.get_select_by_game_server_id_string(game_server_id)}",
) )
results = self._context.select(UserGameIdent.get_select_by_game_server_id_string(game_server_id)) return List(
for result in results: UserGameIdent,
self._logger.trace(__name__, f"Get UserGameIdent with id {result[0]}") [
joins.append(self._from_result(result)) self._from_result(game_ident)
for game_ident in self._context.select(
return joins UserGameIdent.get_select_by_game_server_id_string(game_server_id)
)
],
)
def get_user_game_ident_by_id(self, id: int) -> UserGameIdent: def get_user_game_ident_by_id(self, id: int) -> UserGameIdent:
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {UserGameIdent.get_select_by_id_string(id)}", f"Send SQL command: {UserGameIdent.get_select_by_id_string(id)}",
) )
result = self._context.select(UserGameIdent.get_select_by_id_string(id))[0] return self._from_result(self._context.select(UserGameIdent.get_select_by_id_string(id))[0])
return self._from_result(result)
def get_user_game_ident_by_ident(self, ident: str) -> UserGameIdent: def get_user_game_ident_by_ident(self, ident: str) -> UserGameIdent:
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {UserGameIdent.get_select_by_ident_string(ident)}", f"Send SQL command: {UserGameIdent.get_select_by_ident_string(ident)}",
) )
result = self._context.select(UserGameIdent.get_select_by_ident_string(ident))[0] return self._from_result(self._context.select(UserGameIdent.get_select_by_ident_string(ident))[0])
return self._from_result(result)
def find_user_game_ident_by_ident(self, ident: str) -> Optional[UserGameIdent]: def find_user_game_ident_by_ident(self, ident: str) -> Optional[UserGameIdent]:
self._logger.trace( self._logger.trace(
@ -89,20 +90,20 @@ class UserGameIdentRepositoryService(UserGameIdentRepositoryABC):
if len(result) == 0: if len(result) == 0:
return None return None
result = result[0] return self._from_result(result[0])
return self._from_result(result)
def get_user_game_idents_by_user_id(self, user_id: int) -> List[UserGameIdent]: def get_user_game_idents_by_user_id(self, user_id: int) -> List[UserGameIdent]:
joins = List(UserGameIdent)
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {UserGameIdent.get_select_by_user_id_string(user_id)}", f"Send SQL command: {UserGameIdent.get_select_by_user_id_string(user_id)}",
) )
results = self._context.select(UserGameIdent.get_select_by_user_id_string(user_id)) return List(
for result in results: UserGameIdent,
joins.append(self._from_result(result)) [
self._from_result(game_ident)
return joins for game_ident in self._context.select(UserGameIdent.get_select_by_user_id_string(user_id))
],
)
def add_user_game_ident(self, user_game_ident: UserGameIdent): def add_user_game_ident(self, user_game_ident: UserGameIdent):
self._logger.trace(__name__, f"Send SQL command: {user_game_ident.insert_string}") self._logger.trace(__name__, f"Send SQL command: {user_game_ident.insert_string}")