Added user warning model #35

kdb-bot/src/bot/startup_migration_extension.py

@@ -16,6 +16,7 @@ from bot_data.migration.user_joined_game_server_migration import UserJoinedGameS
 from bot_data.migration.user_message_count_per_hour_migration import (
     UserMessageCountPerHourMigration,
 )
+from bot_data.migration.user_warning_migration import UserWarningMigration
 from bot_data.service.migration_service import MigrationService
 
 
@@ -38,3 +39,4 @@ class StartupMigrationExtension(StartupExtensionABC):
         services.add_transient(MigrationABC, ApiKeyMigration)  # 09.02.2023 #162 - 1.0.0
         services.add_transient(MigrationABC, UserJoinedGameServerMigration)  # 12.02.2023 #181 - 1.0.0
         services.add_transient(MigrationABC, RemoveStatsMigration)  # 19.02.2023 #190 - 1.0.0
+        services.add_transient(MigrationABC, UserWarningMigration)  # 21.02.2023 #35 - 1.0.0

kdb-bot/src/bot_data/migration/user_warning_migration.py

@@ -0,0 +1,35 @@
+from bot_core.logging.database_logger import DatabaseLogger
+from bot_data.abc.migration_abc import MigrationABC
+from bot_data.db_context import DBContext
+
+
+class UserWarningMigration(MigrationABC):
+    name = "1.0_UserWarningMigration"
+
+    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 `UserWarnings` (
+                `Id` BIGINT NOT NULL AUTO_INCREMENT,
+                `Description` VARCHAR(255) NOT NULL,
+                `Author` BIGINT NULL,
+                `CreatedAt` DATETIME(6),
+                `LastModifiedAt` DATETIME(6),
+                PRIMARY KEY(`Id`),
+                FOREIGN KEY (`Author`) REFERENCES `Users`(`UserId`)
+            );
+            """
+            )
+        )
+
+    def downgrade(self):
+        self._cursor.execute("DROP TABLE `UserWarnings`;")

kdb-bot/src/bot_data/model/user_warning.py

@@ -0,0 +1,99 @@
+from datetime import datetime
+from typing import Optional
+
+from cpl_core.database import TableABC
+
+from bot_data.model.user import User
+
+
+class UserWarning(TableABC):
+    def __init__(
+        self,
+        description: str,
+        user: Optional[User],
+        created_at: datetime = None,
+        modified_at: datetime = None,
+        id=0,
+    ):
+        self._id = id
+        self._description = description
+        self._user = user
+        self._author = user
+
+        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 id(self) -> int:
+        return self._id
+
+    @property
+    def description(self) -> str:
+        return self._description
+
+    @property
+    def author(self) -> Optional[User]:
+        return self._author
+
+    @staticmethod
+    def get_select_all_string() -> str:
+        return str(
+            f"""
+                SELECT * FROM `UserWarnings`;
+            """
+        )
+
+    @staticmethod
+    def get_select_by_id_string(id: int) -> str:
+        return str(
+            f"""
+                SELECT * FROM `UserWarnings`
+                WHERE `UserWarnings` = {id};
+            """
+        )
+
+    @staticmethod
+    def get_select_by_user_id_string(id: int) -> str:
+        return str(
+            f"""
+                SELECT * FROM `UserWarnings`
+                WHERE `UserId` = {id};
+            """
+        )
+
+    @property
+    def insert_string(self) -> str:
+        return str(
+            f"""
+            INSERT INTO `UserWarnings` (
+                `Description`, `UserId`, {"" if self._author is None else f"`Author`,"} `CreatedAt`, `LastModifiedAt`
+            ) VALUES (
+                '{self._description}',
+                {self._user.id},
+                {"" if self._author is None else f"{self._author.id},"}
+                '{self._created_at}',
+                '{self._modified_at}'
+            );
+            """
+        )
+
+    @property
+    def udpate_string(self) -> str:
+        return str(
+            f"""
+            UPDATE `UserWarnings`
+            SET `Description` = '{self._description}',
+            `LastModifiedAt` = '{self._modified_at}'
+            WHERE `Id` = {self._id};
+            """
+        )
+
+    @property
+    def delete_string(self) -> str:
+        return str(
+            f"""
+            DELETE FROM `UserWarnings`
+            WHERE `Id` = {self._id};
+            """
+        )