kdb-bot/src/bot/startup_migration_extension.py

@@ -4,6 +4,7 @@ from cpl_core.dependency_injection import ServiceCollectionABC
 from cpl_core.environment import ApplicationEnvironmentABC
 
 from bot_data.abc.migration_abc import MigrationABC
+from bot_data.migration.api_key_migration import ApiKeyMigration
 from bot_data.migration.api_migration import ApiMigration
 from bot_data.migration.auto_role_fix1_migration import AutoRoleFix1Migration
 from bot_data.migration.auto_role_migration import AutoRoleMigration
@@ -32,3 +33,4 @@ class StartupMigrationExtension(StartupExtensionABC):
         services.add_transient(MigrationABC, StatsMigration)  # 09.11.2022 #46 - 0.3.0
         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

kdb-bot/src/bot_data/abc/api_key_repository_abc.py

@@ -0,0 +1,31 @@
+from abc import ABC, abstractmethod
+
+from cpl_query.extension import List
+
+from bot_data.model.api_key import ApiKey
+
+
+class ApiKeyRepositoryABC(ABC):
+    @abstractmethod
+    def __init__(self):
+        pass
+
+    @abstractmethod
+    def get_api_keys(self) -> List[ApiKey]:
+        pass
+
+    @abstractmethod
+    def get_api_key(self, identifier: str, key: str) -> ApiKey:
+        pass
+
+    @abstractmethod
+    def add_api_key(self, api_key: ApiKey):
+        pass
+
+    @abstractmethod
+    def update_api_key(self, api_key: ApiKey):
+        pass
+
+    @abstractmethod
+    def delete_api_key(self, api_key: ApiKey):
+        pass

kdb-bot/src/bot_data/data_module.py

@@ -5,6 +5,7 @@ from cpl_discord.service.discord_collection_abc import DiscordCollectionABC
 
 from bot_core.abc.module_abc import ModuleABC
 from bot_core.configuration.feature_flags_enum import FeatureFlagsEnum
+from bot_data.abc.api_key_repository_abc import ApiKeyRepositoryABC
 from bot_data.abc.auth_user_repository_abc import AuthUserRepositoryABC
 from bot_data.abc.auto_role_repository_abc import AutoRoleRepositoryABC
 from bot_data.abc.client_repository_abc import ClientRepositoryABC
@@ -20,6 +21,7 @@ from bot_data.abc.user_message_count_per_hour_repository_abc import (
     UserMessageCountPerHourRepositoryABC,
 )
 from bot_data.abc.user_repository_abc import UserRepositoryABC
+from bot_data.service.api_key_repository_service import ApiKeyRepositoryService
 from bot_data.service.auth_user_repository_service import AuthUserRepositoryService
 from bot_data.service.auto_role_repository_service import AutoRoleRepositoryService
 from bot_data.service.client_repository_service import ClientRepositoryService
@@ -48,6 +50,7 @@ class DataModule(ModuleABC):
         pass
 
     def configure_services(self, services: ServiceCollectionABC, env: ApplicationEnvironmentABC):
+        services.add_transient(ApiKeyRepositoryABC, ApiKeyRepositoryService)
         services.add_transient(AuthUserRepositoryABC, AuthUserRepositoryService)
         services.add_transient(ServerRepositoryABC, ServerRepositoryService)
         services.add_transient(UserRepositoryABC, UserRepositoryService)

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

@@ -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 ApiKeyMigration(MigrationABC):
+    name = "1.0_ApiKeyMigration"
+
+    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 `ApiKeys` (
+                `Id` BIGINT NOT NULL AUTO_INCREMENT,
+                `Identifier` VARCHAR(255) NOT NULL,
+                `Key` VARCHAR(255) NOT NULL,
+                `CreatorId` BIGINT,
+                `CreatedAt` DATETIME(6),
+                `LastModifiedAt` DATETIME(6),
+                PRIMARY KEY(`Id`),
+                FOREIGN KEY (`CreatorId`) REFERENCES `Users`(`UserId`),
+                CONSTRAINT UC_Identifier_Key UNIQUE (`Identifier`,`Key`)
+            );
+            """
+            )
+        )
+
+    def downgrade(self):
+        self._cursor.execute("DROP TABLE `ApiKeys`;")

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

@@ -0,0 +1,92 @@
+from datetime import datetime
+
+from cpl_core.database import TableABC
+
+from bot_data.model.user import User
+
+
+class ApiKey(TableABC):
+    def __init__(
+        self,
+        identifier: str,
+        key: str,
+        creator: User,
+        created_at: datetime = None,
+        modified_at: datetime = None,
+        id=0,
+    ):
+        self._id = id
+        self._identifier = identifier
+        self._key = key
+        self._creator = creator
+
+        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 identifier(self) -> str:
+        return self._identifier
+
+    @property
+    def key(self) -> str:
+        return self._key
+
+    @property
+    def creator(self) -> User:
+        return self._creator
+
+    @staticmethod
+    def get_select_all_string() -> str:
+        return str(
+            f"""
+            SELECT * FROM `ApiKeys`;
+        """
+        )
+
+    @staticmethod
+    def get_select_string(identifier: str, key: str) -> str:
+        return str(
+            f"""
+            SELECT * FROM `ApiKeys`
+            WHERE `Identifier` = '{identifier}'
+            AND `Key` = '{key}';
+        """
+        )
+
+    @property
+    def insert_string(self) -> str:
+        return str(
+            f"""
+            INSERT INTO `ApiKeys` (
+                `Identifier`, `Key`, `CreatorId`, `CreatedAt`, `LastModifiedAt`
+            ) VALUES (
+                '{self._identifier}',
+                '{self._key}',
+                '{self._creator.user_id}',
+                '{self._created_at}',
+                '{self._modified_at}'
+            );
+        """
+        )
+
+    @property
+    def udpate_string(self) -> str:
+        return str(
+            f"""
+            UPDATE `ApiKeys`
+            SET `Identifier` = '{self._identifier}',
+            `Key` = '{self._key}',
+            `LastModifiedAt` = '{self._modified_at}'
+            WHERE `Id` = {self._id};
+        """
+        )
+
+    @property
+    def delete_string(self) -> str:
+        return str(
+            f"""
+            DELETE FROM `ApiKeys`
+            WHERE `Id` = {self._id};
+        """
+        )

kdb-bot/src/bot_data/service/api_key_repository_service.py

@@ -0,0 +1,73 @@
+from typing import Optional
+
+from cpl_core.database.context import DatabaseContextABC
+from cpl_core.utils import CredentialManager
+from cpl_query.extension import List
+
+from bot_core.logging.database_logger import DatabaseLogger
+from bot_data.abc.api_key_repository_abc import ApiKeyRepositoryABC
+from bot_data.abc.user_repository_abc import UserRepositoryABC
+from bot_data.model.api_key import ApiKey
+
+
+class ApiKeyRepositoryService(ApiKeyRepositoryABC):
+    def __init__(
+        self,
+        logger: DatabaseLogger,
+        db_context: DatabaseContextABC,
+        users: UserRepositoryABC,
+    ):
+        self._logger = logger
+        self._context = db_context
+
+        self._users = users
+
+        ApiKeyRepositoryABC.__init__(self)
+
+    @staticmethod
+    def _get_value_from_result(value: any) -> Optional[any]:
+        if isinstance(value, str) and "NULL" in value:
+            return None
+
+        return value
+
+    def _api_key_from_result(self, sql_result: tuple) -> ApiKey:
+        code = self._get_value_from_result(sql_result[3])
+        if code is not None:
+            code = CredentialManager.decrypt(code)
+
+        api_key = ApiKey(
+            self._get_value_from_result(sql_result[1]),
+            self._get_value_from_result(sql_result[2]),
+            self._users.get_user_by_id(int(self._get_value_from_result(sql_result[3]))),
+            self._get_value_from_result(sql_result[4]),
+            self._get_value_from_result(sql_result[5]),
+            id=self._get_value_from_result(sql_result[0]),
+        )
+
+        return api_key
+
+    def get_api_keys(self) -> List[ApiKey]:
+        api_keys = List(ApiKey)
+        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 api_keys
+
+    def get_api_key(self, identifier: str, key: str) -> ApiKey:
+        self._logger.trace(__name__, f"Send SQL command: {ApiKey.get_select_string(identifier, key)}")
+        return self._get_value_from_result(self._context.select(ApiKey.get_select_string(identifier, key))[0])
+
+    def add_api_key(self, api_key: ApiKey):
+        self._logger.trace(__name__, f"Send SQL command: {api_key.insert_string}")
+        self._context.cursor.execute(api_key.insert_string)
+
+    def update_api_key(self, api_key: ApiKey):
+        self._logger.trace(__name__, f"Send SQL command: {api_key.udpate_string}")
+        self._context.cursor.execute(api_key.udpate_string)
+
+    def delete_api_key(self, api_key: ApiKey):
+        self._logger.trace(__name__, f"Send SQL command: {api_key.delete_string}")
+        self._context.cursor.execute(api_key.delete_string)