Compare commits

..

No commits in common. "0aa690b98486d4017c499d8922c20cc899f1c61e" and "a2dd447dbdc339d871857aac0c668feaae4385c9" have entirely different histories.

20 changed files with 646 additions and 363 deletions

View File

@ -48,60 +48,66 @@ 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 List( return achievements
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)}")
return self._from_result(self._context.select(Achievement.get_select_by_id_string(id))[0]) 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 List( return achievements
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))
return List( for achievements_join in achievements_joins:
UserGotAchievement, results = self._context.select(Achievement.get_select_by_id_string(achievements_join.achievement.id))
[ for result in results:
self._join_from_result(result).achievement self._logger.trace(__name__, f"Got Achievement with id {result[0]}")
for result in self._context.select(UserGotAchievement.get_select_by_user_id_string(user_id)) achievements.append(self._from_result(result))
],
) 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 List( return achievements_joins
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,12 +44,13 @@ 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 List( return api_keys
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,16 +64,23 @@ 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 List(int, [result[0] for result in self._context.select(auth_user.get_select_user_id_from_relations())]) return relation_ids
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 List( return users
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()
@ -122,7 +129,8 @@ 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)}")
return self._user_from_result(self._context.select(AuthUser.get_select_by_email_string(email))[0]) 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)}")
@ -130,7 +138,9 @@ class AuthUserRepositoryService(AuthUserRepositoryABC):
if result is None or len(result) == 0: if result is None or len(result) == 0:
return None return None
return self._user_from_result(result[0]) 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(
@ -141,7 +151,9 @@ class AuthUserRepositoryService(AuthUserRepositoryABC):
if result is None or len(result) == 0: if result is None or len(result) == 0:
return None return None
return self._user_from_result(result[0]) 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(
@ -152,7 +164,9 @@ class AuthUserRepositoryService(AuthUserRepositoryABC):
if result is None or len(result) == 0: if result is None or len(result) == 0:
return None return None
return self._user_from_result(result[0]) 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,7 +23,27 @@ class AutoRoleRepositoryService(AutoRoleRepositoryABC):
AutoRoleRepositoryABC.__init__(self) AutoRoleRepositoryABC.__init__(self)
def _from_result(self, result: tuple) -> AutoRole: def get_auto_roles(self) -> List[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],
@ -33,37 +53,55 @@ 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
return self._from_result(result[0]) 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)}")
return List( auto_roles = List(AutoRole)
AutoRole, results = self._context.select(AutoRole.get_select_by_server_id_string(id))
[self._from_result(result) for result in self._context.select(AutoRole.get_select_by_server_id_string(id))], 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_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)}",
) )
return self._from_result(self._context.select(AutoRole.get_select_by_message_id_string(id))[0]) 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(
@ -74,7 +112,16 @@ class AutoRoleRepositoryService(AutoRoleRepositoryABC):
if result is None or len(result) == 0: if result is None or len(result) == 0:
return None return None
return self._from_result(result[0]) 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}")
@ -88,7 +135,27 @@ 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 _rule_from_result(self, result: tuple) -> AutoRoleRule: def get_auto_role_rules(self) -> List[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],
@ -98,29 +165,26 @@ 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)}",
) )
return List( results = self._context.select(AutoRoleRule.get_select_by_auto_role_id_string(id))
AutoRoleRule, for result in results:
[ auto_role_rules.append(
self._rule_from_result(result) AutoRoleRule(
for result in self._context.select(AutoRoleRule.get_select_by_auto_role_id_string(id)) self.get_auto_role_by_id(result[1]),
], 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,7 +23,32 @@ class ClientRepositoryService(ClientRepositoryABC):
ClientRepositoryABC.__init__(self) ClientRepositoryABC.__init__(self)
def _from_result(self, result: tuple) -> Client: def get_clients(self) -> List[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],
@ -37,37 +62,49 @@ 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))
return List( for result in results:
Client, clients.append(
[ Client(
self._from_result(result) result[1],
for result in self._context.select(Client.get_select_by_server_id_string(server_id)) 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_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)}",
) )
return self._from_result(self._context.select(Client.get_select_by_discord_id_string(discord_id))[0]) 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(
@ -78,7 +115,20 @@ class ClientRepositoryService(ClientRepositoryABC):
if result is None or len(result) == 0: if result is None or len(result) == 0:
return None return None
return self._from_result(result[0]) 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(
@ -89,7 +139,20 @@ class ClientRepositoryService(ClientRepositoryABC):
if result is None or len(result) == 0: if result is None or len(result) == 0:
return None return None
return self._from_result(result[0]) 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(
@ -100,7 +163,20 @@ class ClientRepositoryService(ClientRepositoryABC):
if result is None or len(result) == 0: if result is None or len(result) == 0:
return None return None
return self._from_result(result[0]) 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 List( return game_servers
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 List( return game_servers
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)}",
) )
return self._from_result(self._context.select(GameServer.get_select_by_id_string(id))[0]) 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)}",
) )
return List( game_servers = List(GameServer)
GameServer, results = self._context.select(GameServer.get_select_by_api_key_id_string(id))
[ for result in results:
self._from_result(result) game_servers.append(self._from_result(result))
for result in self._context.select(GameServer.get_select_by_api_key_id_string(id)) return game_servers
],
)
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.level_repository_abc import LevelRepositoryABC
from bot_data.abc.server_repository_abc import ServerRepositoryABC from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.abc.level_repository_abc import LevelRepositoryABC
from bot_data.model.level import Level from bot_data.model.level import Level
@ -41,15 +41,20 @@ 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()}")
return List( results = self._context.select(Level.get_select_all_string())
Level, [self._level_from_result(result) for result in self._context.select(Level.get_select_all_string())] for result in results:
) 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(self._context.select(Level.get_select_by_id_string(id))[0]) return self._level_from_result(result)
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)}")
@ -60,19 +65,21 @@ 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)}",
) )
return List( results = self._context.select(Level.get_select_by_server_id_string(server_id))
Level,
[ for result in results:
self._level_from_result(result) self._logger.trace(__name__, f"Get level with id {result[0]}")
for result in self._context.select(Level.get_select_by_server_id_string(server_id)) levels.append(self._level_from_result(result))
],
) 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)}",
@ -81,10 +88,11 @@ class LevelRepositoryService(LevelRepositoryABC):
if results is None or len(results) == 0: if results is None or len(results) == 0:
return None return None
return List( for result in results:
Level, self._logger.trace(__name__, f"Get level with id {result[0]}")
[self._level_from_result(result) for result in results], levels.append(self._level_from_result(result))
)
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.scheduled_event_repository_abc import ScheduledEventRepositoryABC
from bot_data.abc.server_repository_abc import ServerRepositoryABC from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.abc.scheduled_event_repository_abc import ScheduledEventRepositoryABC
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 List( return scheduled_events
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(self._context.select(ScheduledEvent.get_select_by_id_string(id))[0]) return self._scheduled_event_from_result(result)
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))
return List( for result in results:
ScheduledEvent, self._logger.trace(__name__, f"Get scheduled_event with id {result[0]}")
[ scheduled_events.append(self._scheduled_event_from_result(result))
self._scheduled_event_from_result(result)
for result in self._context.select(ScheduledEvent.get_select_by_server_id_string(server_id)) return scheduled_events
],
)
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}")

View File

@ -26,14 +26,15 @@ 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))
return List( for result in results:
ServerTeamRoleIdsConfig, self._logger.trace(__name__, f"Got ServerTeamRoleIdsConfig with id {result[0]}")
[ ids.append(
ServerTeamRoleIdsConfig( ServerTeamRoleIdsConfig(
result[1], result[1],
TeamMemberTypeEnum(result[2]), TeamMemberTypeEnum(result[2]),
@ -42,23 +43,22 @@ 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 List( return urls
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,14 +102,18 @@ 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)}",
) )
return self._from_result(self._context.select(ServerConfig.get_select_by_server_id_string(server_id))[0]) 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)}",
) )
return self._from_result(self._context.select(ServerConfig.get_select_by_id_string(config_id))[0]) 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,15 +25,12 @@ 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
@ -99,7 +96,9 @@ class ServerRepositoryService(ServerRepositoryABC):
if result is None or len(result) == 0: if result is None or len(result) == 0:
return None return None
return Server(result[0][1], result[0][2], result[0][3], id=result[0][0]) result = result[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,15 +43,14 @@ 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 List( return short_role_names
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)}")
@ -60,30 +59,31 @@ 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)}",
) )
return List( results = self._context.select(ShortRoleName.get_select_by_role_id_string(role_id))
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_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)}",
) )
return List( results = self._context.select(ShortRoleName.get_select_by_server_id_string(server_id))
ShortRoleName,
[ for result in results:
self._short_role_name_from_result(result) self._logger.trace(__name__, f"Get short_role_name with id {result[0]}")
for result in self._context.select(ShortRoleName.get_select_by_server_id_string(server_id)) short_role_names.append(self._short_role_name_from_result(result))
],
) 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,24 +42,23 @@ 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 List( return steam_special_offers
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)}",
) )
return self._steam_special_offer_from_result( result = self._context.select(SteamSpecialOffer.get_select_by_name_string(name))[0]
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,17 +18,27 @@ 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()}")
return List(int, [config[1] for config in self._context.select(TechnicianIdConfig.get_select_all_string())]) results = 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()}",
) )
return List( results = self._context.select(TechnicianPingUrlConfig.get_select_all_string())
str, [ping_url[1] for ping_url in self._context.select(TechnicianPingUrlConfig.get_select_all_string())] for result in results:
) 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(
@ -54,7 +64,9 @@ 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()}")
return self._from_result(self._context.select(TechnicianConfig.get_select_all_string())[0]) 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,47 +39,46 @@ 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 List( return joins
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)}",
) )
return List( results = self._context.select(UserGameIdent.get_select_by_game_server_id_string(game_server_id))
UserGameIdent, for result in results:
[ self._logger.trace(__name__, f"Get UserGameIdent with id {result[0]}")
self._from_result(game_ident) joins.append(self._from_result(result))
for game_ident in self._context.select(
UserGameIdent.get_select_by_game_server_id_string(game_server_id) return joins
)
],
)
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)}",
) )
return self._from_result(self._context.select(UserGameIdent.get_select_by_id_string(id))[0]) 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)}",
) )
return self._from_result(self._context.select(UserGameIdent.get_select_by_ident_string(ident))[0]) 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(
@ -90,20 +89,20 @@ class UserGameIdentRepositoryService(UserGameIdentRepositoryABC):
if len(result) == 0: if len(result) == 0:
return None return None
return self._from_result(result[0]) 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)}",
) )
return List( results = self._context.select(UserGameIdent.get_select_by_user_id_string(user_id))
UserGameIdent, for result in results:
[ joins.append(self._from_result(result))
self._from_result(game_ident)
for game_ident in self._context.select(UserGameIdent.get_select_by_user_id_string(user_id)) return joins
],
)
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}")

View File

@ -39,45 +39,45 @@ class UserJoinedGameServerRepositoryService(UserJoinedGameServerRepositoryABC):
) )
def get_user_joined_game_servers(self) -> List[UserJoinedGameServer]: def get_user_joined_game_servers(self) -> List[UserJoinedGameServer]:
joins = List(UserJoinedGameServer)
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {UserJoinedGameServer.get_select_all_string()}", f"Send SQL command: {UserJoinedGameServer.get_select_all_string()}",
) )
results = self._context.select(UserJoinedGameServer.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Get user-joined-game-server with id {result[0]}")
joins.append(self._from_result(result))
return List( return joins
UserJoinedGameServer,
[self._from_result(join) for join in self._context.select(UserJoinedGameServer.get_select_all_string())],
)
def get_user_joined_game_server_by_id(self, id: int) -> UserJoinedGameServer: def get_user_joined_game_server_by_id(self, id: int) -> UserJoinedGameServer:
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {UserJoinedGameServer.get_select_by_id_string(id)}", f"Send SQL command: {UserJoinedGameServer.get_select_by_id_string(id)}",
) )
return self._from_result(self._context.select(UserJoinedGameServer.get_select_by_id_string(id))[0]) result = self._context.select(UserJoinedGameServer.get_select_by_id_string(id))[0]
return self._from_result(result)
def get_user_joined_game_servers_by_user_id(self, user_id: int) -> List[UserJoinedGameServer]: def get_user_joined_game_servers_by_user_id(self, user_id: int) -> List[UserJoinedGameServer]:
joins = List(UserJoinedGameServer)
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {UserJoinedGameServer.get_select_by_user_id_string(user_id)}", f"Send SQL command: {UserJoinedGameServer.get_select_by_user_id_string(user_id)}",
) )
results = self._context.select(UserJoinedGameServer.get_select_by_user_id_string(user_id))
for result in results:
joins.append(self._from_result(result))
return List( return joins
UserJoinedGameServer,
[
self._from_result(join)
for join in self._context.select(UserJoinedGameServer.get_select_by_user_id_string(user_id))
],
)
def get_active_user_joined_game_server_by_user_id(self, user_id: int) -> UserJoinedGameServer: def get_active_user_joined_game_server_by_user_id(self, user_id: int) -> UserJoinedGameServer:
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {UserJoinedGameServer.get_select_by_user_id_string(user_id)}", f"Send SQL command: {UserJoinedGameServer.get_select_by_user_id_string(user_id)}",
) )
return self._from_result( result = self._context.select(UserJoinedGameServer.get_select_active_by_user_id_string(user_id))[0]
self._context.select(UserJoinedGameServer.get_select_active_by_user_id_string(user_id))[0] return self._from_result(result)
)
def find_active_user_joined_game_server_by_user_id(self, user_id: int) -> Optional[UserJoinedGameServer]: def find_active_user_joined_game_server_by_user_id(self, user_id: int) -> Optional[UserJoinedGameServer]:
self._logger.trace( self._logger.trace(
@ -88,21 +88,22 @@ class UserJoinedGameServerRepositoryService(UserJoinedGameServerRepositoryABC):
if result is None or len(result) == 0: if result is None or len(result) == 0:
return None return None
return self._from_result(result[0]) result = result[0]
return self._from_result(result)
def find_active_user_joined_game_servers_by_user_id(self, user_id: int) -> List[Optional[UserJoinedGameServer]]: def find_active_user_joined_game_servers_by_user_id(self, user_id: int) -> List[Optional[UserJoinedGameServer]]:
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {UserJoinedGameServer.get_select_active_by_user_id_string(user_id)}", f"Send SQL command: {UserJoinedGameServer.get_select_active_by_user_id_string(user_id)}",
) )
result = List(UserJoinedGameServer)
db_results = self._context.select(UserJoinedGameServer.get_select_active_by_user_id_string(user_id))
return List( for db_result in db_results:
UserJoinedGameServer, result.append(self._from_result(db_result))
[
self._from_result(join) return result
for join in self._context.select(UserJoinedGameServer.get_select_active_by_user_id_string(user_id))
],
)
def add_user_joined_game_server(self, user_joined_game_server: UserJoinedGameServer): def add_user_joined_game_server(self, user_joined_game_server: UserJoinedGameServer):
self._logger.trace(__name__, f"Send SQL command: {user_joined_game_server.insert_string}") self._logger.trace(__name__, f"Send SQL command: {user_joined_game_server.insert_string}")

View File

@ -23,7 +23,31 @@ class UserJoinedServerRepositoryService(UserJoinedServerRepositoryABC):
UserJoinedServerRepositoryABC.__init__(self) UserJoinedServerRepositoryABC.__init__(self)
def __from_result(self, result: tuple) -> UserJoinedServer: def get_user_joined_servers(self) -> List[UserJoinedServer]:
joins = List(UserJoinedServer)
self._logger.trace(__name__, f"Send SQL command: {UserJoinedServer.get_select_all_string()}")
results = self._context.select(UserJoinedServer.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Get user-joined-server with id {result[0]}")
joins.append(
UserJoinedServer(
self._users.get_user_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
)
return joins
def get_user_joined_server_by_id(self, id: int) -> UserJoinedServer:
self._logger.trace(
__name__,
f"Send SQL command: {UserJoinedServer.get_select_by_id_string(id)}",
)
result = self._context.select(UserJoinedServer.get_select_by_id_string(id))[0]
return UserJoinedServer( return UserJoinedServer(
self._users.get_user_by_id(result[1]), self._users.get_user_by_id(result[1]),
result[2], result[2],
@ -33,48 +57,55 @@ class UserJoinedServerRepositoryService(UserJoinedServerRepositoryABC):
id=result[0], id=result[0],
) )
def get_user_joined_servers(self) -> List[UserJoinedServer]:
self._logger.trace(__name__, f"Send SQL command: {UserJoinedServer.get_select_all_string()}")
return List(
UserJoinedServer,
[self.__from_result(join) for join in self._context.select(UserJoinedServer.get_select_all_string())],
)
def get_user_joined_server_by_id(self, id: int) -> UserJoinedServer:
self._logger.trace(
__name__,
f"Send SQL command: {UserJoinedServer.get_select_by_id_string(id)}",
)
return self.__from_result(self._context.select(UserJoinedServer.get_select_by_id_string(id))[0])
def get_user_joined_server_by_server_id(self, server_id: int) -> UserJoinedServer: def get_user_joined_server_by_server_id(self, server_id: int) -> UserJoinedServer:
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {UserJoinedServer.get_select_by_server_id_string(server_id)}", f"Send SQL command: {UserJoinedServer.get(id)}",
)
result = self._context.select(UserJoinedServer.get_select_by_id_string(id))[0]
return UserJoinedServer(
self._users.get_user_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
) )
return self.__from_result(self._context.select(UserJoinedServer.get_select_by_id_string(server_id))[0])
def get_user_joined_servers_by_user_id(self, user_id: int) -> List[UserJoinedServer]: def get_user_joined_servers_by_user_id(self, user_id: int) -> List[UserJoinedServer]:
joins = List(UserJoinedServer)
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {UserJoinedServer.get_select_by_user_id_string(user_id)}", f"Send SQL command: {UserJoinedServer.get_select_by_user_id_string(user_id)}",
) )
return List( results = self._context.select(UserJoinedServer.get_select_by_user_id_string(user_id))
UserJoinedServer, for result in results:
[ joins.append(
self.__from_result(join) UserJoinedServer(
for join in self._context.select(UserJoinedServer.get_select_by_user_id_string(user_id)) self._users.get_user_by_id(result[1]),
], result[2],
result[3],
result[4],
result[5],
id=result[0],
) )
)
return joins
def get_active_user_joined_server_by_user_id(self, user_id: int) -> UserJoinedServer: def get_active_user_joined_server_by_user_id(self, user_id: int) -> UserJoinedServer:
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {UserJoinedServer.get_select_by_user_id_string(user_id)}", f"Send SQL command: {UserJoinedServer.get_select_by_user_id_string(user_id)}",
) )
return self.__from_result( result = self._context.select(UserJoinedServer.get_select_active_by_user_id_string(user_id))[0]
self._context.select(UserJoinedServer.get_select_active_by_user_id_string(user_id))[0] return UserJoinedServer(
self._users.get_user_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
) )
def find_active_user_joined_server_by_user_id(self, user_id: int) -> Optional[UserJoinedServer]: def find_active_user_joined_server_by_user_id(self, user_id: int) -> Optional[UserJoinedServer]:
@ -86,7 +117,16 @@ class UserJoinedServerRepositoryService(UserJoinedServerRepositoryABC):
if result is None or len(result) == 0: if result is None or len(result) == 0:
return None return None
return self.__from_result(result[0]) result = result[0]
return UserJoinedServer(
self._users.get_user_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
def add_user_joined_server(self, user_joined_server: UserJoinedServer): def add_user_joined_server(self, user_joined_server: UserJoinedServer):
self._logger.trace(__name__, f"Send SQL command: {user_joined_server.insert_string}") self._logger.trace(__name__, f"Send SQL command: {user_joined_server.insert_string}")

View File

@ -25,7 +25,34 @@ class UserJoinedVoiceChannelRepositoryService(UserJoinedVoiceChannelRepositoryAB
UserJoinedVoiceChannelRepositoryABC.__init__(self) UserJoinedVoiceChannelRepositoryABC.__init__(self)
def __from_result(self, result: tuple) -> UserJoinedVoiceChannel: def get_user_joined_voice_channels(self) -> List[UserJoinedVoiceChannel]:
joins = List(UserJoinedVoiceChannel)
self._logger.trace(
__name__,
f"Send SQL command: {UserJoinedVoiceChannel.get_select_all_string()}",
)
results = self._context.select(UserJoinedVoiceChannel.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Get user-joined-voice-channel with id {result[0]}")
joins.append(
UserJoinedVoiceChannel(
self._users.get_user_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
)
return joins
def get_user_joined_voice_channel_by_id(self, id: int) -> UserJoinedVoiceChannel:
self._logger.trace(
__name__,
f"Send SQL command: {UserJoinedVoiceChannel.get_select_by_id_string(id)}",
)
result = self._context.select(UserJoinedVoiceChannel.get_select_by_id_string(id))[0]
return UserJoinedVoiceChannel( return UserJoinedVoiceChannel(
self._users.get_user_by_id(result[1]), self._users.get_user_by_id(result[1]),
result[2], result[2],
@ -35,45 +62,40 @@ class UserJoinedVoiceChannelRepositoryService(UserJoinedVoiceChannelRepositoryAB
id=result[0], id=result[0],
) )
def get_user_joined_voice_channels(self) -> List[UserJoinedVoiceChannel]:
self._logger.trace(
__name__,
f"Send SQL command: {UserJoinedVoiceChannel.get_select_all_string()}",
)
return List(
UserJoinedVoiceChannel,
[self.__from_result(join) for join in self._context.select(UserJoinedVoiceChannel.get_select_all_string())],
)
def get_user_joined_voice_channel_by_id(self, id: int) -> UserJoinedVoiceChannel:
self._logger.trace(
__name__,
f"Send SQL command: {UserJoinedVoiceChannel.get_select_by_id_string(id)}",
)
return self.__from_result(self._context.select(UserJoinedVoiceChannel.get_select_by_id_string(id))[0])
def get_user_joined_voice_channels_by_user_id(self, user_id: int) -> List[UserJoinedVoiceChannel]: def get_user_joined_voice_channels_by_user_id(self, user_id: int) -> List[UserJoinedVoiceChannel]:
joins = List(UserJoinedVoiceChannel)
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {UserJoinedVoiceChannel.get_select_by_user_id_string(user_id)}", f"Send SQL command: {UserJoinedVoiceChannel.get_select_by_user_id_string(user_id)}",
) )
results = self._context.select(UserJoinedVoiceChannel.get_select_by_user_id_string(user_id))
return List( for result in results:
UserJoinedVoiceChannel, joins.append(
[ UserJoinedVoiceChannel(
self.__from_result(join) self._users.get_user_by_id(result[1]),
for join in self._context.select(UserJoinedVoiceChannel.get_select_by_user_id_string(user_id)) result[2],
], result[3],
result[4],
result[5],
id=result[0],
) )
)
return joins
def get_active_user_joined_voice_channel_by_user_id(self, user_id: int) -> UserJoinedVoiceChannel: def get_active_user_joined_voice_channel_by_user_id(self, user_id: int) -> UserJoinedVoiceChannel:
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {UserJoinedVoiceChannel.get_select_by_user_id_string(user_id)}", f"Send SQL command: {UserJoinedVoiceChannel.get_select_by_user_id_string(user_id)}",
) )
return self.__from_result( result = self._context.select(UserJoinedVoiceChannel.get_select_active_by_user_id_string(user_id))[0]
self._context.select(UserJoinedVoiceChannel.get_select_active_by_user_id_string(user_id))[0] return UserJoinedVoiceChannel(
self._users.get_user_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
) )
def find_active_user_joined_voice_channel_by_user_id(self, user_id: int) -> Optional[UserJoinedVoiceChannel]: def find_active_user_joined_voice_channel_by_user_id(self, user_id: int) -> Optional[UserJoinedVoiceChannel]:
@ -85,21 +107,38 @@ class UserJoinedVoiceChannelRepositoryService(UserJoinedVoiceChannelRepositoryAB
if result is None or len(result) == 0: if result is None or len(result) == 0:
return None return None
return self.__from_result(result[0]) result = result[0]
return UserJoinedVoiceChannel(
self._users.get_user_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
def find_active_user_joined_voice_channels_by_user_id(self, user_id: int) -> List[Optional[UserJoinedVoiceChannel]]: def find_active_user_joined_voice_channels_by_user_id(self, user_id: int) -> List[Optional[UserJoinedVoiceChannel]]:
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {UserJoinedVoiceChannel.get_select_active_by_user_id_string(user_id)}", f"Send SQL command: {UserJoinedVoiceChannel.get_select_active_by_user_id_string(user_id)}",
) )
result = List(UserJoinedVoiceChannel)
db_results = self._context.select(UserJoinedVoiceChannel.get_select_active_by_user_id_string(user_id))
return List( for db_result in db_results:
UserJoinedVoiceChannel, result.append(
[ UserJoinedVoiceChannel(
self.__from_result(join) self._users.get_user_by_id(db_result[1]),
for join in self._context.select(UserJoinedVoiceChannel.get_select_active_by_user_id_string(user_id)) db_result[2],
], db_result[3],
db_result[4],
db_result[5],
id=db_result[0],
) )
)
return result
def add_user_joined_voice_channel(self, user_joined_voice_channel: UserJoinedVoiceChannel): def add_user_joined_voice_channel(self, user_joined_voice_channel: UserJoinedVoiceChannel):
self._logger.trace(__name__, f"Send SQL command: {user_joined_voice_channel.insert_string}") self._logger.trace(__name__, f"Send SQL command: {user_joined_voice_channel.insert_string}")

View File

@ -44,30 +44,39 @@ class UserMessageCountPerHourRepositoryService(UserMessageCountPerHourRepository
) )
def get_user_message_count_per_hours(self) -> List[UserMessageCountPerHour]: def get_user_message_count_per_hours(self) -> List[UserMessageCountPerHour]:
umcphs = List(UserMessageCountPerHour)
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {UserMessageCountPerHour.get_select_all_string()}", f"Send SQL command: {UserMessageCountPerHour.get_select_all_string()}",
) )
results = self._context.select(UserMessageCountPerHour.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Get user message count per hour with id {result[0]}")
umcphs.append(self._from_result(result))
return List( return umcphs
UserMessageCountPerHour,
[
self._from_result(umcphs)
for umcphs in self._context.select(UserMessageCountPerHour.get_select_all_string())
],
)
def find_user_message_count_per_hour_by_user_id(self, user_id: int) -> List[Optional[UserMessageCountPerHour]]: def find_user_message_count_per_hour_by_user_id(self, user_id: int) -> List[Optional[UserMessageCountPerHour]]:
umcphs = List(UserMessageCountPerHour)
sql = UserMessageCountPerHour.get_select_by_user_id_string(user_id) sql = UserMessageCountPerHour.get_select_by_user_id_string(user_id)
self._logger.trace(__name__, f"Send SQL command: {sql}") self._logger.trace(__name__, f"Send SQL command: {sql}")
return List(UserMessageCountPerHour, [self._from_result(umcphs) for umcphs in self._context.select(sql)]) results = self._context.select(sql)
if results is None or len(results) == 0:
return umcphs
for result in results:
self._logger.trace(__name__, f"Get user message count per hour with id {result[0]}")
umcphs.append(self._from_result(result))
return umcphs
def get_user_message_count_per_hour_by_user_id_and_date( def get_user_message_count_per_hour_by_user_id_and_date(
self, user_id: int, date: datetime self, user_id: int, date: datetime
) -> UserMessageCountPerHour: ) -> UserMessageCountPerHour:
sql = UserMessageCountPerHour.get_select_by_user_id_and_date_string(user_id, date) sql = UserMessageCountPerHour.get_select_by_user_id_and_date_string(user_id, date)
self._logger.trace(__name__, f"Send SQL command: {sql}") self._logger.trace(__name__, f"Send SQL command: {sql}")
return self._from_result(self._context.select(sql)[0]) result = self._context.select(sql)[0]
return self._from_result(result)
def find_user_message_count_per_hour_by_user_id_and_date( def find_user_message_count_per_hour_by_user_id_and_date(
self, user_id: int, date: datetime self, user_id: int, date: datetime

View File

@ -1,3 +1,4 @@
import datetime
from typing import Optional from typing import Optional
from cpl_core.database.context import DatabaseContextABC from cpl_core.database.context import DatabaseContextABC
@ -37,8 +38,14 @@ class UserRepositoryService(UserRepositoryABC):
) )
def get_users(self) -> List[User]: def get_users(self) -> List[User]:
users = List(User)
self._logger.trace(__name__, f"Send SQL command: {User.get_select_all_string()}") self._logger.trace(__name__, f"Send SQL command: {User.get_select_all_string()}")
return List(User, [self._from_result(user) for user in self._context.select(User.get_select_all_string())]) results = self._context.select(User.get_select_all_string())
for result in results:
self._logger.trace(__name__, f"Get user with id {result[0]}")
users.append(self._from_result(result))
return users
def get_user_by_id(self, id: int) -> User: def get_user_by_id(self, id: int) -> User:
self._logger.trace(__name__, f"Send SQL command: {User.get_select_by_id_string(id)}") self._logger.trace(__name__, f"Send SQL command: {User.get_select_by_id_string(id)}")
@ -52,40 +59,42 @@ class UserRepositoryService(UserRepositoryABC):
if result is None or len(result) == 0: if result is None or len(result) == 0:
return None return None
return self._from_result(result[0]) result = result[0]
return self._from_result(result)
def get_users_by_discord_id(self, discord_id: int) -> List[User]: def get_users_by_discord_id(self, discord_id: int) -> List[User]:
users = List(User)
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {User.get_select_by_discord_id_string(discord_id)}", f"Send SQL command: {User.get_select_by_discord_id_string(discord_id)}",
) )
return List( results = self._context.select(User.get_select_by_discord_id_string(discord_id))
User, for result in results:
[ users.append(self._from_result(result))
self._from_result(user)
for user in self._context.select(User.get_select_by_discord_id_string(discord_id)) return users
],
)
def get_users_by_server_id(self, server_id: int) -> List[User]: def get_users_by_server_id(self, server_id: int) -> List[User]:
users = List(User)
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {User.get_select_by_server_id_string(server_id)}", f"Send SQL command: {User.get_select_by_server_id_string(server_id)}",
) )
return List( results = self._context.select(User.get_select_by_server_id_string(server_id))
User, for result in results:
[self._from_result(user) for user in self._context.select(User.get_select_by_server_id_string(server_id))], users.append(self._from_result(result))
)
return users
def get_user_by_discord_id_and_server_id(self, discord_id: int, server_id: int) -> User: def get_user_by_discord_id_and_server_id(self, discord_id: int, server_id: int) -> User:
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {User.get_select_by_discord_id_and_server_id_string(discord_id, server_id)}", f"Send SQL command: {User.get_select_by_discord_id_and_server_id_string(discord_id, server_id)}",
) )
result = self._context.select(User.get_select_by_discord_id_and_server_id_string(discord_id, server_id))[0]
return self._from_result( return self._from_result(result)
self._context.select(User.get_select_by_discord_id_and_server_id_string(discord_id, server_id))[0]
)
def find_user_by_discord_id_and_server_id(self, discord_id: int, server_id: int) -> Optional[User]: def find_user_by_discord_id_and_server_id(self, discord_id: int, server_id: int) -> Optional[User]:
self._logger.trace( self._logger.trace(
@ -96,7 +105,9 @@ class UserRepositoryService(UserRepositoryABC):
if result is None or len(result) == 0: if result is None or len(result) == 0:
return None return None
return self._from_result(result[0]) result = result[0]
return self._from_result(result)
def add_user(self, user: User): def add_user(self, user: User):
self._logger.trace(__name__, f"Send SQL command: {user.insert_string}") self._logger.trace(__name__, f"Send SQL command: {user.insert_string}")

View File

@ -46,29 +46,30 @@ class UserWarningsRepositoryService(UserWarningsRepositoryABC):
) )
def get_user_warnings(self) -> List[UserWarnings]: def get_user_warnings(self) -> List[UserWarnings]:
warnings = List(UserWarnings)
self._logger.trace(__name__, f"Send SQL command: {UserWarnings.get_select_all_string()}") self._logger.trace(__name__, f"Send SQL command: {UserWarnings.get_select_all_string()}")
return List( results = self._context.select(UserWarnings.get_select_all_string())
UserWarnings, for result in results:
[self._from_result(warning) for warning in self._context.select(UserWarnings.get_select_all_string())], warnings.append(self._from_result(result))
)
return warnings
def get_user_warnings_by_id(self, id: int) -> UserWarnings: def get_user_warnings_by_id(self, id: int) -> UserWarnings:
self._logger.trace(__name__, f"Send SQL command: {UserWarnings.get_select_by_id_string(id)}") self._logger.trace(__name__, f"Send SQL command: {UserWarnings.get_select_by_id_string(id)}")
return self._from_result(self._context.select(UserWarnings.get_select_by_id_string(id))[0]) result = self._context.select(UserWarnings.get_select_by_id_string(id))[0]
return self._from_result(result)
def get_user_warnings_by_user_id(self, user_id: int) -> List[UserWarnings]: def get_user_warnings_by_user_id(self, user_id: int) -> List[UserWarnings]:
warnings = List(UserWarnings)
self._logger.trace( self._logger.trace(
__name__, __name__,
f"Send SQL command: {UserWarnings.get_select_by_user_id_string(user_id)}", f"Send SQL command: {UserWarnings.get_select_by_user_id_string(user_id)}",
) )
results = self._context.select(UserWarnings.get_select_by_user_id_string(user_id))
for result in results:
warnings.append(self._from_result(result))
return List( return warnings
UserWarnings,
[
self._from_result(warning)
for warning in self._context.select(UserWarnings.get_select_by_user_id_string(user_id))
],
)
def add_user_warnings(self, user_warnings: UserWarnings): def add_user_warnings(self, user_warnings: UserWarnings):
self._logger.trace(__name__, f"Send SQL command: {user_warnings.insert_string}") self._logger.trace(__name__, f"Send SQL command: {user_warnings.insert_string}")