72 lines
2.6 KiB
Python
72 lines
2.6 KiB
Python
from typing import Optional
|
|
from cpl_core.database.context import DatabaseContextABC
|
|
from cpl_core.logging import LoggerABC
|
|
from cpl_query.extension import List
|
|
|
|
from gismo_data.abc.server_repository_abc import ServerRepositoryABC
|
|
from gismo_data.model.server import Server
|
|
|
|
|
|
class ServerRepositoryService(ServerRepositoryABC):
|
|
|
|
def __init__(self, logger: LoggerABC, db_context: DatabaseContextABC):
|
|
self._logger = logger
|
|
self._context = db_context
|
|
|
|
ServerRepositoryABC.__init__(self)
|
|
|
|
def get_servers(self) -> List[Server]:
|
|
servers = List(Server)
|
|
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],
|
|
id=result[0]
|
|
))
|
|
|
|
return servers
|
|
|
|
def get_server_by_id(self, id: int) -> Server:
|
|
self._logger.trace(__name__, f'Send SQL command: {Server.get_select_by_id_string(id)}')
|
|
result = self._context.select(Server.get_select_by_id_string(id))[0]
|
|
return Server(
|
|
result[1],
|
|
id=result[0]
|
|
)
|
|
|
|
def get_server_by_discord_id(self, discord_id: int) -> Server:
|
|
self._logger.trace(__name__, f'Send SQL command: {Server.get_select_by_discord_id_string(discord_id)}')
|
|
result = self._context.select(Server.get_select_by_discord_id_string(discord_id))
|
|
return Server(
|
|
result[1],
|
|
id=result[0]
|
|
)
|
|
|
|
def find_server_by_discord_id(self, discord_id: int) -> Optional[Server]:
|
|
self._logger.trace(__name__, f'Send SQL command: {Server.get_select_by_discord_id_string(discord_id)}')
|
|
result = self._context.select(Server.get_select_by_discord_id_string(discord_id))
|
|
if result is None or len(result) == 0:
|
|
return None
|
|
|
|
result = result[0]
|
|
|
|
return Server(
|
|
result[1],
|
|
result[2],
|
|
result[3],
|
|
id=result[0]
|
|
)
|
|
|
|
def add_server(self, server: Server):
|
|
self._logger.trace(__name__, f'Send SQL command: {server.insert_string}')
|
|
self._context.cursor.execute(server.insert_string)
|
|
|
|
def update_server(self, server: Server):
|
|
self._logger.trace(__name__, f'Send SQL command: {server.udpate_string}')
|
|
self._context.cursor.execute(server.udpate_string)
|
|
|
|
def delete_server(self, server: Server):
|
|
self._logger.trace(__name__, f'Send SQL command: {server.delete_string}')
|
|
self._context.cursor.execute(server.delete_string)
|