src/liquidcode/application.py - develop-0.2.0 - liquid-g/liquid-code

src/liquidcode/application.py
319 строк13 KB
User
0.3.0-RC1 - обработка дефолтного роута.
04 июл 2026, 18:34
04 июл 2026, 18:34a9918be
"""
Основной класс приложения LiquidCode (фасад).
Объединяет DI-контейнер, роутер, ядро, статику, шаблоны и конфигурацию.
"""

import inspect
import logging
import os
from typing import Any, Callable, Optional, Type, List

from . import __version__
from .config import load_config, ConfigError
from .container import Container
from .discovery import discover_controllers, discover_services
from .kernel import Kernel
from .routing import Router
from .server import HTTPServerManager
from .template import TemplateManager

logger = logging.getLogger(__name__)


class Application:
    """
    Фасад для удобной настройки и запуска приложения.
    """

    def __init__(
            self,
            container: Optional[Container] = None,
            response_class: Optional[Type] = None,
            profiler_enabled: bool = False,
            enable_service_discovery: bool = True,
            enable_controller_discovery: bool = True,
    ) -> None:
        self.container = container or Container()
        self.router = Router()
        self.kernel = Kernel(self.router, self.container, response_class, profiler_enabled)

        # Настройки по умолчанию
        self.host = 'localhost'
        self.port = 8000
        self._static_prefix = '/static'
        self._static_dir = 'public'
        self._template_dir = 'templates'
        self._debug = False

        # Шаблоны
        self.template_manager = TemplateManager(self._template_dir)
        # Регистрируем TemplateManager в контейнере для внедрения в контроллеры
        self.container.set(TemplateManager, self.template_manager)

        self._config_loaded = False

        # Авто-дискавери сервисов и контроллеров
        if enable_service_discovery:
            self.discover_services()
        if enable_controller_discovery:
            self.discover_controllers()

    # ---------- Переменные окружения ----------

    def load_env(self, file_path: str = '.env') -> None:
        from .env import load_env as load_env_file
        load_env_file(file_path)

    # ---------- Конфигурация ----------

    def load_config(self, file_path: str) -> None:
        if self._config_loaded:
            logger.warning("Конфиг уже загружен. Повторная загрузка пропущена.")
            return

        try:
            config = load_config(file_path)
        except ConfigError as e:
            logger.error(f"Ошибка загрузки конфига: {e}")
            return

        config = self._interpolate_config(config)
        params = config.get('parameters', {})

        if 'host' in params:
            self.host = params['host']
        if 'port' in params:
            self.port = int(params['port'])
        if 'static_prefix' in params:
            self._static_prefix = params['static_prefix']
        if 'static_dir' in params:
            self._static_dir = params['static_dir']
        if 'template_dir' in params:
            self._template_dir = params['template_dir']
            self.template_manager = TemplateManager(self._template_dir)
            self.container.set(TemplateManager, self.template_manager)
        if 'debug' in params:
            self._debug = bool(params['debug'])
        if 'middleware_config' in params:
            self.load_middleware_config(params['middleware_config'])
        elif 'middleware' in config:
            self._load_middleware_from_config(config['middleware'])

        self._config_loaded = True
        logger.info(f"Конфиг загружен из {file_path}")

    def _interpolate_config(self, value):
        if isinstance(value, dict):
            return {k: self._interpolate_config(v) for k, v in value.items()}
        elif isinstance(value, list):
            return [self._interpolate_config(v) for v in value]
        elif isinstance(value, str):
            return os.path.expandvars(value)
        else:
            return value

    def _load_middleware_from_config(self, middleware_list: list) -> None:
        import importlib
        items = sorted(middleware_list, key=lambda x: x.get('priority', 0))
        for item in items:
            if not item.get('enabled', True):
                continue
            class_path = item['class']
            try:
                module_name, class_name = class_path.rsplit('.', 1)
                module = importlib.import_module(module_name)
                middleware_class = getattr(module, class_name)
                # Проверяем, принимает ли конструктор параметры
                sig = inspect.signature(middleware_class.__init__)
                params = {}
                if 'container' in sig.parameters:
                    params['container'] = self.container
                if 'debug' in sig.parameters:
                    params['debug'] = self._debug
                instance = middleware_class(**params)
                self.add_middleware(instance)
                logger.info(f"Загружен middleware {class_name} из конфига")
            except Exception as e:
                logger.error(f"Ошибка загрузки middleware {class_path}: {e}")

    # ---------- Шаблоны ----------

    def render_template(self, template_name: str, **context: Any) -> str:
        return self.template_manager.render(template_name, **context)

    # ---------- Статика ----------

    def set_static(self, prefix: str = '/static', directory: str = 'public') -> None:
        self._static_prefix = prefix.rstrip('/')
        self._static_dir = directory
        logger.info(f"Статика из '{directory}' доступна по '{prefix}'")

    # ---------- Контроллеры ----------

    def add_controller(self, controller_class: Type[Any]) -> None:
        self.router.add_routes_from_controller(controller_class)
        logger.info(f"Зарегистрирован контроллер {controller_class.__name__}")

    def discover_controllers(self, directory: str = 'src/Controller') -> None:
        discover_controllers(directory, self.add_controller)

    # ---------- Сервисы ----------

    def discover_services(self, directory: str = 'src/Service') -> None:
        discover_services(directory, self.container)

    # ---------- Маршруты ----------

    def route(self, path: str, methods: Optional[List[str]] = None):
        def decorator(func):
            self.router.add_route(path, func, methods or ['GET'])
            return func

        return decorator

    def websocket_route(self, path: str):
        def decorator(func):
            self.router.add_route(path, func, ['WEBSOCKET'])
            return func

        return decorator

    # ---------- Middleware ----------

    def add_middleware(self, middleware: Callable) -> None:
        self.kernel.add_middleware(middleware)

    def group(self, prefix: str, middlewares: Optional[List[Callable]] = None):
        return self.router.group(prefix, middlewares)

    def load_middleware_config(self, file_path: str = 'config/middleware.yaml') -> None:
        import importlib
        from .config import load_config

        try:
            config = load_config(file_path)
        except ConfigError as e:
            logger.warning(f"Не удалось загрузить конфиг middleware: {e}")
            return

        items = config.get('middleware', [])
        items.sort(key=lambda x: x.get('priority', 0))

        enabled_count = sum(1 for item in items if item.get('enabled', True))
        logger.info(f"Найдено {len(items)} middleware (из них включено: {enabled_count})")

        for item in items:
            if not item.get('enabled', True):
                logger.debug(f"Пропущен отключенный middleware {item['class']}")
                continue

            class_path = item['class']
            try:
                module_name, class_name = class_path.rsplit('.', 1)
                module = importlib.import_module(module_name)
                middleware_class = getattr(module, class_name)
                # Проверяем, принимает ли конструктор параметры
                sig = inspect.signature(middleware_class.__init__)
                params = {}
                if 'container' in sig.parameters:
                    params['container'] = self.container
                if 'debug' in sig.parameters:
                    params['debug'] = self._debug
                instance = middleware_class(**params)
                self.add_middleware(instance)
                logger.info(f"Загружен middleware {class_name} из конфига")
            except Exception as e:
                logger.error(f"Ошибка загрузки middleware {class_path}: {e}")

    # ---------- Хуки ----------

    def add_before_request(self, hook: Callable) -> None:
        self.kernel.add_before_request(hook)

    def add_after_request(self, hook: Callable) -> None:
        self.kernel.add_after_request(hook)

    def add_exception(self, hook: Callable) -> None:
        self.kernel.add_exception(hook)

    def add_terminate(self, hook: Callable) -> None:
        self.kernel.add_terminate(hook)

    def enable_profiler(self) -> None:
        """Включает профилировщик."""
        self.kernel.profiler_enabled = True
        logger.info("Профилировщик включён")

    # ---------- Запуск серверов ----------

    def run(self, host: Optional[str] = None, port: Optional[int] = None) -> None:
        self._ensure_logging()
        self._ensure_default_route()

        manager = HTTPServerManager(self.kernel, self.host, self.port)
        manager.set_debug(self._debug)
        manager.run(host, port)

    def run_websocket(self, host: Optional[str] = None, port: Optional[int] = None) -> None:
        """Запускает WebSocket-сервер (требуется установить websockets)."""
        from .server.websocket import WebSocketServerManager
        manager = WebSocketServerManager(self.kernel, self.host, 8765)
        manager.run(host, port)

    # ---------- Вспомогательные ----------

    def _ensure_default_route(self) -> None:
        if self.router.has_route('/', 'GET'):
            return

        def default_handler(**kwargs):
            try:
                html = self.render_template('welcome.html.j2', version=__version__)
                return html.encode('utf-8'), 200, {'Content-Type': 'text/html'}
            except Exception:
                html = f"""
                <!DOCTYPE html>
                <html>
                <head>
                    <meta charset="UTF-8">
                    <title>LiquidCode</title>
                    <style>
                        body {{ font-family: system-ui, sans-serif; text-align: center; padding: 50px; }}
                        h1 {{ font-size: 3rem; }}
                        code {{ background: #f0f0f0; padding: 2px 6px; border-radius: 4px; }}
                    </style>
                </head>
                <body>
                    <h1>🚀 LiquidCode</h1>
                    <p>Версия: <strong>{__version__}</strong></p>
                    <p>Создайте шаблон <code>templates/welcome.html.j2</code> для кастомизации этой страницы.</p>
                    <p><a href="/routes">Посмотреть маршруты</a></p>
                </body>
                </html>
                """
                return html.encode('utf-8'), 200, {'Content-Type': 'text/html'}

        self.set_default_route(default_handler)

    def set_default_route(self, handler: Callable) -> None:
        if self.router.has_route('/', 'GET'):
            logger.warning("Дефолтный маршрут уже существует, будет перезаписан.")

        def wrapped(**kwargs):
            result = handler(**kwargs)
            if isinstance(result, tuple) and len(result) == 3:
                return result
            return result

        self.router.add_route('/', wrapped, ['GET'])

    def _ensure_logging(self) -> None:
        root_logger = logging.getLogger()
        if not root_logger.handlers:
            handler = logging.StreamHandler()
            level = logging.DEBUG if self._debug else logging.WARNING
            handler.setLevel(level)
            formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
            handler.setFormatter(formatter)
            root_logger.addHandler(handler)
            root_logger.setLevel(level)