sh-edraft.de/sh_discord_bot
4
1
Fork 1
Code Issues 21 Pull Requests Actions Packages Projects 3 Releases 27 Wiki Activity

Added user joined game server model #181

Browse Source
This commit is contained in:
Sven Heidemann 2023-02-12 15:42:02 +01:00
parent 91bbb2f2e9
commit 0b5acdea58
6 changed files with 408 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
kdb-bot/src/bot/startup_migration_extension.py
View File

@ -11,6 +11,7 @@ from bot_data.migration.auto_role_migration import AutoRoleMigration
from bot_data.migration.initial_migration import InitialMigration
from bot_data.migration.level_migration import LevelMigration
from bot_data.migration.stats_migration import StatsMigration
from bot_data.migration.user_joined_game_server_migration import UserJoinedGameServerMigration
from bot_data.migration.user_message_count_per_hour_migration import (
UserMessageCountPerHourMigration,
)
@ -34,3 +35,4 @@ class StartupMigrationExtension(StartupExtensionABC):
services.add_transient(MigrationABC, AutoRoleFix1Migration) # 30.12.2022 #151 - 0.3.0
services.add_transient(MigrationABC, UserMessageCountPerHourMigration) # 11.01.2023 #168 - 0.3.1
services.add_transient(MigrationABC, ApiKeyMigration) # 09.02.2023 #162 - 1.0.0
services.add_transient(MigrationABC, UserJoinedGameServerMigration) # 12.02.2023 #181 - 1.0.0

52
kdb-bot/src/bot_data/abc/user_joined_game_server_repository_abc.py Normal file
View File

@ -0,0 +1,52 @@
from abc import ABC, abstractmethod
from typing import Optional
from cpl_query.extension import List
from bot_data.model.user_joined_game_server import UserJoinedGameServer
class UserJoinedGameServerRepositoryABC(ABC):
@abstractmethod
def __init__(self):
pass
@abstractmethod
def get_user_joined_game_servers(self) -> List[UserJoinedGameServer]:
pass
@abstractmethod
def get_user_joined_game_server_by_id(self, id: int) -> UserJoinedGameServer:
pass
@abstractmethod
def get_user_joined_game_servers_by_user_id(self, user_id: int) -> List[UserJoinedGameServer]:
pass
@abstractmethod
def get_active_user_joined_game_server_by_user_id(self, user_id: int) -> UserJoinedGameServer:
pass
@abstractmethod
def find_active_user_joined_game_server_by_user_id(self, user_id: int) -> Optional[UserJoinedGameServer]:
pass
@abstractmethod
def find_active_user_joined_game_servers_by_user_id(self, user_id: int) -> List[Optional[UserJoinedGameServer]]:
pass
@abstractmethod
def add_user_joined_game_server(self, user_joined_game_server: UserJoinedGameServer):
pass
@abstractmethod
def update_user_joined_game_server(self, user_joined_game_server: UserJoinedGameServer):
pass
@abstractmethod
def delete_user_joined_game_server(self, user_joined_game_server: UserJoinedGameServer):
pass
@abstractmethod
def delete_user_joined_game_server_by_user_id(self, user_id: int):
pass

3
kdb-bot/src/bot_data/data_module.py
View File

@ -13,6 +13,7 @@ from bot_data.abc.known_user_repository_abc import KnownUserRepositoryABC
from bot_data.abc.level_repository_abc import LevelRepositoryABC
from bot_data.abc.server_repository_abc import ServerRepositoryABC
from bot_data.abc.statistic_repository_abc import StatisticRepositoryABC
from bot_data.abc.user_joined_game_server_repository_abc import UserJoinedGameServerRepositoryABC
from bot_data.abc.user_joined_server_repository_abc import UserJoinedServerRepositoryABC
from bot_data.abc.user_joined_voice_channel_repository_abc import (
UserJoinedVoiceChannelRepositoryABC,
@ -30,6 +31,7 @@ from bot_data.service.level_repository_service import LevelRepositoryService
from bot_data.service.seeder_service import SeederService
from bot_data.service.server_repository_service import ServerRepositoryService
from bot_data.service.statistic_repository_service import StatisticRepositoryService
from bot_data.service.user_joined_game_server_repository_service import UserJoinedGameServerRepositoryService
from bot_data.service.user_joined_server_repository_service import (
UserJoinedServerRepositoryService,
)
@ -58,6 +60,7 @@ class DataModule(ModuleABC):
services.add_transient(KnownUserRepositoryABC, KnownUserRepositoryService)
services.add_transient(UserJoinedServerRepositoryABC, UserJoinedServerRepositoryService)
services.add_transient(UserJoinedVoiceChannelRepositoryABC, UserJoinedVoiceChannelRepositoryService)
services.add_transient(UserJoinedGameServerRepositoryABC, UserJoinedGameServerRepositoryService)
services.add_transient(AutoRoleRepositoryABC, AutoRoleRepositoryService)
services.add_transient(LevelRepositoryABC, LevelRepositoryService)
services.add_transient(StatisticRepositoryABC, StatisticRepositoryService)

37
kdb-bot/src/bot_data/migration/user_joined_game_server_migration.py Normal file
View File

@ -0,0 +1,37 @@
from bot_core.logging.database_logger import DatabaseLogger
from bot_data.abc.migration_abc import MigrationABC
from bot_data.db_context import DBContext
class UserJoinedGameServerMigration(MigrationABC):
name = "1.0_UserJoinedGameServerMigration"
def __init__(self, logger: DatabaseLogger, db: DBContext):
MigrationABC.__init__(self)
self._logger = logger
self._db = db
self._cursor = db.cursor
def upgrade(self):
self._logger.debug(__name__, "Running upgrade")
self._cursor.execute(
str(
f"""
CREATE TABLE IF NOT EXISTS `UserJoinedGameServer` (
`Id` BIGINT NOT NULL AUTO_INCREMENT,
`UserId` BIGINT NOT NULL,
`GameServer` VARCHAR(255) NOT NULL,
`JoinedOn` DATETIME(6) NOT NULL,
`LeavedOn` DATETIME(6),
`CreatedAt` DATETIME(6),
`LastModifiedAt` DATETIME(6),
FOREIGN KEY (`UserId`) REFERENCES Users(`UserId`),
PRIMARY KEY(`Id`)
);
"""
)
)
def downgrade(self):
self._cursor.execute("DROP TABLE `UserJoinedGameServer`;")

154
kdb-bot/src/bot_data/model/user_joined_game_server.py Normal file
View File

@ -0,0 +1,154 @@
from datetime import datetime
from cpl_core.database import TableABC
from bot_data.model.user import User
class UserJoinedGameServer(TableABC):
def __init__(
self,
user: User,
game_server: str,
joined_on: datetime,
leaved_on: datetime = None,
created_at: datetime = None,
modified_at: datetime = None,
id=0,
):
self._join_id = id
self._game_server = game_server
self._user = user
self._joined_on = joined_on
self._leaved_on = leaved_on
TableABC.__init__(self)
self._created_at = created_at if created_at is not None else self._created_at
self._modified_at = modified_at if modified_at is not None else self._modified_at
@property
def join_id(self) -> int:
return self._join_id
@property
def game_server(self) -> str:
return self._game_server
@property
def user(self) -> User:
return self._user
@property
def joined_on(self) -> datetime:
return self._joined_on
@joined_on.setter
def joined_on(self, value: datetime):
self._modified_at = datetime.now()
self.joined_on = value
@property
def leaved_on(self) -> datetime:
return self._leaved_on
@leaved_on.setter
def leaved_on(self, value: datetime):
self._modified_at = datetime.now()
self._leaved_on = value
@staticmethod
def get_select_all_string() -> str:
return str(
f"""
SELECT * FROM `UserJoinedGameServer`;
"""
)
@staticmethod
def get_select_by_id_string(id: int) -> str:
return str(
f"""
SELECT * FROM `UserJoinedGameServer`
WHERE `Id` = {id};
"""
)
@staticmethod
def get_select_by_user_id_string(id: int) -> str:
return str(
f"""
SELECT * FROM `UserJoinedGameServer`
WHERE `UserId` = {id};
"""
)
@staticmethod
def get_select_active_by_user_id_string(id: int) -> str:
return str(
f"""
SELECT * FROM `UserJoinedGameServer`
WHERE `UserId` = {id}
AND `LeavedOn` IS NULL;
"""
)
@property
def insert_string(self) -> str:
if self._leaved_on is not None:
return str(
f"""
INSERT INTO `UserJoinedGameServer` (
`UserId`, `GameServer`, `JoinedOn`, `LeavedOn`, `CreatedAt`, `LastModifiedAt`
) VALUES (
{self._user.user_id},
{self._game_server},
'{self._joined_on}',
'{self._leaved_on}',
'{self._created_at}',
'{self._modified_at}'
);
"""
)
else:
return str(
f"""
INSERT INTO `UserJoinedGameServer` (
`UserId`, `GameServer`, `JoinedOn`, `CreatedAt`, `LastModifiedAt`
) VALUES (
{self._user.user_id},
{self._game_server},
'{self._joined_on}',
'{self._created_at}',
'{self._modified_at}'
);
"""
)
@property
def udpate_string(self) -> str:
return str(
f"""
UPDATE `UserJoinedGameServer`
SET `LeavedOn` = '{self._leaved_on}',
`LastModifiedAt` = '{self._modified_at}'
WHERE `Id` = {self._join_id};
"""
)
@property
def delete_string(self) -> str:
return str(
f"""
DELETE FROM `UserJoinedGameServer`
WHERE `Id` = {self._join_id};
"""
)
@staticmethod
def delete_by_user_id_string(id: int) -> str:
return str(
f"""
DELETE FROM `UserJoinedGameServer`
WHERE `UserId` = {id}
"""
)

160
kdb-bot/src/bot_data/service/user_joined_game_server_repository_service.py Normal file
View File

@ -0,0 +1,160 @@
from typing import Optional
from cpl_core.database.context import DatabaseContextABC
from cpl_query.extension import List
from bot_core.logging.database_logger import DatabaseLogger
from bot_data.abc.user_joined_game_server_repository_abc import (
UserJoinedGameServerRepositoryABC,
)
from bot_data.abc.user_repository_abc import UserRepositoryABC
from bot_data.model.user_joined_game_server import UserJoinedGameServer
class UserJoinedGameServerRepositoryService(UserJoinedGameServerRepositoryABC):
def __init__(
self,
logger: DatabaseLogger,
db_context: DatabaseContextABC,
users: UserRepositoryABC,
):
self._logger = logger
self._context = db_context
self._users = users
UserJoinedGameServerRepositoryABC.__init__(self)
def get_user_joined_game_servers(self) -> List[UserJoinedGameServer]:
joins = List(UserJoinedGameServer)
self._logger.trace(
__name__,
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(
UserJoinedGameServer(
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_game_server_by_id(self, id: int) -> UserJoinedGameServer:
self._logger.trace(
__name__,
f"Send SQL command: {UserJoinedGameServer.get_select_by_id_string(id)}",
)
result = self._context.select(UserJoinedGameServer.get_select_by_id_string(id))[0]
return UserJoinedGameServer(
self._users.get_user_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
def get_user_joined_game_servers_by_user_id(self, user_id: int) -> List[UserJoinedGameServer]:
joins = List(UserJoinedGameServer)
self._logger.trace(
__name__,
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(
UserJoinedGameServer(
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_game_server_by_user_id(self, user_id: int) -> UserJoinedGameServer:
self._logger.trace(
__name__,
f"Send SQL command: {UserJoinedGameServer.get_select_by_user_id_string(user_id)}",
)
result = self._context.select(UserJoinedGameServer.get_select_active_by_user_id_string(user_id))[0]
return UserJoinedGameServer(
self._users.get_user_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
def find_active_user_joined_game_server_by_user_id(self, user_id: int) -> Optional[UserJoinedGameServer]:
self._logger.trace(
__name__,
f"Send SQL command: {UserJoinedGameServer.get_select_by_user_id_string(user_id)}",
)
result = self._context.select(UserJoinedGameServer.get_select_active_by_user_id_string(user_id))
if result is None or len(result) == 0:
return None
result = result[0]
return UserJoinedGameServer(
self._users.get_user_by_id(result[1]),
result[2],
result[3],
result[4],
result[5],
id=result[0],
)
def find_active_user_joined_game_servers_by_user_id(self, user_id: int) -> List[Optional[UserJoinedGameServer]]:
self._logger.trace(
__name__,
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))
for db_result in db_results:
result.append(
UserJoinedGameServer(
self._users.get_user_by_id(db_result[1]),
db_result[2],
db_result[3],
db_result[4],
db_result[5],
id=db_result[0],
)
)
return result
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._context.cursor.execute(user_joined_game_server.insert_string)
def update_user_joined_game_server(self, user_joined_game_server: UserJoinedGameServer):
self._logger.trace(__name__, f"Send SQL command: {user_joined_game_server.udpate_string}")
self._context.cursor.execute(user_joined_game_server.udpate_string)
def delete_user_joined_game_server(self, user_joined_game_server: UserJoinedGameServer):
self._logger.trace(__name__, f"Send SQL command: {user_joined_game_server.delete_string}")
self._context.cursor.execute(user_joined_game_server.delete_string)
def delete_user_joined_game_server_by_user_id(self, user_id: int):
self._logger.trace(
__name__,
f"Send SQL command: {UserJoinedGameServer.delete_by_user_id_string}",
)
self._context.cursor.execute(UserJoinedGameServer.delete_by_user_id_string(user_id))