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

src/liquidcode/discovery.py
121 строка5 KB
User
0.3.0-RC - дописаны тесты и мелкие исправления
04 июл 2026, 13:07
04 июл 2026, 13:0740ad8b8
"""
Автоматическое обнаружение контроллеров, сервисов и других классов в проекте.
"""

import importlib
import importlib.util
import inspect
import logging
import os
from typing import Callable, Type

from .service import Service

logger = logging.getLogger(__name__)


def has_route_methods(cls: Type) -> bool:
    """
    Проверяет, есть ли у класса методы с атрибутами _route_path и _route_methods.

    Args:
        cls: Класс для проверки.

    Returns:
        True, если у класса есть хотя бы один метод с @route, иначе False.
    """
    for _, method in inspect.getmembers(cls, predicate=inspect.isfunction):
        if hasattr(method, '_route_path') and hasattr(method, '_route_methods'):
            return True
    return False


def discover_classes(
        directory: str,
        predicate: Callable[[Type], bool],
        handler: Callable[[Type], None],
        name: str = 'classes'
) -> None:
    """
    Универсальный поиск классов в директории.

    Args:
        directory: Путь к папке (например, 'src/Controller').
        predicate: Функция, проверяющая, является ли класс подходящим.
        handler: Функция, вызываемая для каждого найденного класса.
        name: Имя для логов (например, 'controllers', 'services').
    """
    if not os.path.exists(directory):
        logger.warning(f"Directory {directory} does not exist. Skipping.")
        return

    # Используем os.walk для надежного обхода файлов
    for root, dirs, files in os.walk(directory):
        # Пропускаем __pycache__
        dirs[:] = [d for d in dirs if d != '__pycache__']
        for file in files:
            if file.endswith('.py') and not file.startswith('__'):
                full_path = os.path.join(root, file)
                # Создаем уникальное имя модуля на основе абсолютного пути
                # Используем tempfile для генерации безопасного имени
                mod_name = f"_discovered_{os.path.basename(root)}_{file[:-3]}"
                try:
                    # Загружаем модуль напрямую по пути
                    spec = importlib.util.spec_from_file_location(mod_name, full_path)
                    if spec is None or spec.loader is None:
                        continue
                    module = importlib.util.module_from_spec(spec)
                    # Добавляем в sys.modules чтобы избежать проблем с относительными импортами
                    import sys
                    sys.modules[mod_name] = module
                    spec.loader.exec_module(module)
                    for _, obj in inspect.getmembers(module, inspect.isclass):
                        # Проверяем, что класс определен в этом модуле, не импортирован
                        if obj.__module__ == mod_name and predicate(obj):
                            handler(obj)
                            logger.info(f"Registered {name[:-1]}: {obj.__name__}")
                except (ImportError, SyntaxError, TypeError) as e:
                    logger.warning(f"Could not import {full_path}: {e}")


def discover_controllers(directory: str, add_controller_func: Callable[[Type], None]) -> None:
    """
    Автоматически находит и регистрирует все контроллеры в указанной директории.

    Args:
        directory: Путь к папке с контроллерами (например, 'src/Controller').
        add_controller_func: Функция, которая принимает класс контроллера для регистрации.
    """
    discover_classes(
        directory=directory,
        predicate=has_route_methods,
        handler=add_controller_func,
        name='controllers'
    )


def discover_services(directory: str, container) -> None:
    """
    Автоматически находит и регистрирует все сервисы в указанной директории.

    Args:
        directory: Путь к папке с сервисами (например, 'src/Service').
        container: DI-контейнер для регистрации сервисов.
    """

    def is_service_class(cls: Type) -> bool:
        return (
                inspect.isclass(cls) and
                issubclass(cls, Service) and
                cls is not Service
        )

    def register_service(cls: Type) -> None:
        container.set(cls, container.get(cls))

    discover_classes(
        directory=directory,
        predicate=is_service_class,
        handler=register_service,
        name='services'
    )