Исходный код EJIO.utils.component.app.app
## /utils/component/app/app.py
import atexit
import os
import signal
import sys
from collections.abc import Callable
from pathlib import Path
from typing import Any
from loguru import logger
from ...interface import Singleton, DisposableMeta
from ...serialization import load
from ...system import import_module_from_path, get_app_root
from .err import ApplicationCoreError
__all__ = [
'ApplicationCore',
]
[документация]
class ApplicationCore(Singleton):
"""
Глобальный оркестратор жизненного цикла всего приложения.
Отвечает за атомарную загрузку конфигураций, динамическую инициализацию,
потокобезопасный перехват сигналов выхода ОС и обработку критических ошибок.
"""
__slots__ = (
'_run_callback',
'_exit_callback',
'_debug',
'_set_logger',
'_app_root',
'_is_cleaned_up',
'_is_started',
)
# Переменная класса для хранения считанных рантайм-настроек проекта
settings: dict[str, Any] = {
'application' : {
'name' : 'DefaultEJIOApp'
}
}
def __init__(
self,
run_callback: Callable[[], None],
exit_callback: Callable[[], None] | None = None,
*,
debug: bool = False,
set_logger: bool = False,
) -> None:
"""
Глобальный оркестратор жизненного цикла всего приложения.
Отвечает за атомарную загрузку конфигураций, динамическую инициализацию,
потокобезопасный перехват сигналов выхода ОС и обработку критических ошибок.
:param run_callback: Главная функция запуска логики (точка входа приложения).
:param exit_callback: Опциональная функция, гарантированно выполняемая при выходе.
:param debug: Если True, включается режим отладки (логирование DEBUG), иначе INFO.
:param set_logger: Если True, устанавливает новый логгер заместо стандартного.
"""
self._run_callback: Callable[[], None] = run_callback
self._exit_callback: Callable[[], None] | None = exit_callback
self._debug: bool = debug
self._set_logger: bool = set_logger
# Определяем физический корень запуска проекта (учитывает компиляцию в .exe)
self._app_root: Path = get_app_root()
self._is_cleaned_up: bool = False
# Флаг защиты от повторного вызова start()
self._is_started: bool = False
# 1. Настраиваем логирование в зависимости от режима отладки
self._configure_logging()
# 2. Регистрируем глобальные ловушки прерывания процессов ОС
self._register_exit_hooks()
def _configure_logging(self) -> None:
"""Переключает детализацию вывода loguru (DEBUG для разработчиков, INFO для релиза)."""
if not self._set_logger:
logger.info("Skipping setting custom logger")
return
logger.info("Setting custom logger")
logger.remove()
log_level = "DEBUG" if self._debug else "INFO"
logger.add(
sys.stderr,
level=log_level,
format=(
"<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | "
"<level>{level: <8}</level> | "
"<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - "
"<level>{message}</level>"
)
)
logger.info(f"Logger core is running in mode: [{log_level}]")
def _register_exit_hooks(self) -> None:
"""Жестко перехватывает любые сценарии закрытия процесса на уровне ядра ОС и sys.exit()."""
# Ловушка 1: Перехват штатного завершения Python и вызовов sys.exit()
atexit.register(self._cleanup)
# Ловушка 2: Перехват системных сигналов прерывания (Ctrl+C, закрытие терминала)
def signal_handler(signum: int, frame: Any) -> None:
logger.warning(f"Got process interrupt system signal: {signum}")
# Вызов sys.exit спровоцирует каскадный переход в atexit-хук
sys.exit(0)
# Привязываем к стандартным сигналам POSIX/Windows
signal.signal(signal.SIGINT, signal_handler) # Ctrl+C
signal.signal(signal.SIGTERM, signal_handler) # Убийство процесса
def _load_settings_file(self) -> None:
"""Сканирует стартовую директорию и автоматически считывает первый найденный файл конфигурации."""
# Возможные имена и расширения конфигурационных файлов в порядке приоритета
supported_configs = ["config.toml", "config.yaml", "config.yml", "config.json", "config.ini"]
for config_name in supported_configs:
potential_path = self._app_root.joinpath(config_name)
if potential_path.exists() and potential_path.is_file():
try:
logger.info(f"Found config file: '{config_name}'. Starting decomposition...")
self.settings = load(None, str(potential_path))
logger.info("Global project settings successfully imported in core memory.")
return
except Exception as e:
logger.warning(f"Failed to parse settings file {config_name}: {e}. Searching for alternatives...")
logger.info("Config files (toml/yaml/json/ini) not found. Core will continue working with empty dictionary.")
def _execute_init_module(self) -> None:
"""Ищет и принудительно выполняет файл init.py в корне перед стартом основной логики."""
init_script_path = self._app_root.joinpath("init.py")
if init_script_path.exists() and init_script_path.is_file():
logger.info("Found preload module 'init.py'. Starting first order initialization...")
try:
import_module_from_path(init_script_path)
logger.info("Module 'init.py' successfully loaded.")
except Exception as e:
logger.critical(
f"Critical error initializing 'init.py': {e}. The application has been terminated abnormally.")
os._exit(1)
else:
logger.info("Missing preload module 'init.py'. Skipping pre-initialization.")
[документация]
def start(self) -> None:
"""
Запускает полный жизненный цикл оркестрации приложения.
:raises ApplicationCoreError: При попытке запуска ядра приложения повторно.
"""
if self._is_started:
if not self._debug:
logger.warning("Blocking: Method ApplicationCore.start() was already started.")
return
logger.critical(f"Blocking: Method ApplicationCore.start() is already started.")
raise ApplicationCoreError('Blocking: Method ApplicationCore.start() is already started.')
self._is_started = True
logger.info("Starting ApplicationCore boot sequence...")
self._execute_init_module()
self._load_settings_file()
logger.info("Boot sequence is completed. Handing control to run_callback().")
try:
self._run_callback()
except Exception as e:
logger.critical(f"Critical unhandled exception in main execution cycle: {e}")
sys.exit(1)
finally:
sys.exit(0)
def _cleanup(self) -> None:
"""Внутренний гарантированный метод очистки, вызываемый при ЛЮБОМ выходе из программы."""
# Защита от двойного вызова деструкции (если atexit и сигналы сработали одновременно)
if self._is_cleaned_up:
return
self._is_cleaned_up = True
logger.info("Initializing application exit procedure...")
if self._exit_callback is not None:
try:
logger.info("Calling user exit function exit_callback()...")
self._exit_callback()
logger.info("User exit function successfully completed.")
except Exception as e:
logger.error(f"Error executing user exit_callback: {e}")
try:
logger.info("Starting clearing Disposable structures)...")
DisposableMeta.dispose_all()
logger.info("Clearing of Disposable structures successfully completed.")
except Exception as e:
logger.error("Error clearing Disposable structures.")
logger.info("ApplicationCore successfully stopped. OS process is completed.")