sh_discord_bot/kdb-bot/src/bot/application.py

from cpl_core.configuration import ConfigurationABC
from cpl_core.console import Console
from cpl_core.dependency_injection import ServiceProviderABC
from cpl_core.logging import LoggerABC
from cpl_discord.application import DiscordBotApplicationABC
from cpl_discord.configuration import DiscordBotSettings
from cpl_discord.service import DiscordBotServiceABC, DiscordBotService
from cpl_translation import TranslatePipe, TranslationServiceABC, TranslationSettings

from bot_api.api_thread import ApiThread
from bot_core.configuration.feature_flags_enum import FeatureFlagsEnum
from bot_core.configuration.feature_flags_settings import FeatureFlagsSettings
from bot_core.service.data_integrity_service import DataIntegrityService


class Application(DiscordBotApplicationABC):
    def __init__(self, config: ConfigurationABC, services: ServiceProviderABC):
        DiscordBotApplicationABC.__init__(self, config, services)

        self._services = services
        self._config = config

        # cpl-core
        self._logger: LoggerABC = services.get_service(LoggerABC)
        self._data_integrity: DataIntegrityService = services.get_service(DataIntegrityService)
        # cpl-discord
        self._bot: DiscordBotServiceABC = services.get_service(DiscordBotServiceABC)
        self._bot_settings: DiscordBotSettings = config.get_configuration(DiscordBotSettings)
        # cpl-translation
        self._translation: TranslationServiceABC = services.get_service(TranslationServiceABC)
        self._t: TranslatePipe = services.get_service(TranslatePipe)

        self._feature_flags: FeatureFlagsSettings = config.get_configuration(FeatureFlagsSettings)

        # api
        if self._feature_flags.get_flag(FeatureFlagsEnum.api_module):
            self._api: ApiThread = services.get_service(ApiThread)

        self._is_stopping = False

    async def configure(self):
        self._translation.load_by_settings(self._configuration.get_configuration(TranslationSettings))

    async def main(self):
        try:
            self._logger.debug(__name__, f"Starting...")

            if (
                self._feature_flags.get_flag(FeatureFlagsEnum.api_module)
                and self._feature_flags.get_flag(FeatureFlagsEnum.api_only)
                and self._environment.environment_name == "development"
            ):
                self._api.start()
                self._api.join()
                return

            self._logger.info(__name__, f"Try to start {DiscordBotService.__name__}")
            await self._bot.start_async()
            await self._bot.stop_async()
        except Exception as e:
            self._logger.error(__name__, "Start failed", e)

    async def stop_async(self):
        if self._is_stopping:
            return

        self._is_stopping = True
        try:
            self._logger.info(__name__, f"Try to stop {DiscordBotService.__name__}")
            if self._feature_flags.get_flag(FeatureFlagsEnum.api_module):
                self._api.stop()

            self._data_integrity.check_data_integrity(is_for_shutdown=True)
            await self._bot.close()
            self._logger.info(__name__, f"Stopped {DiscordBotService.__name__}")
        except Exception as e:
            self._logger.error(__name__, "stop failed", e)

        Console.write_line()

    def is_restart(self):
        return True if self._configuration.get_configuration("IS_RESTART") == "true" else False  #