src/liquidcode/container.py - master - liquid-g/liquid-code

src/liquidcode/container.py
247 строк10 KB
User
0.3.0-RC - дописаны тесты и мелкие исправления
04 июл 2026, 13:07
04 июл 2026, 13:0740ad8b8
"""
DI-контейнер с автопроводкой (autowiring) для LiquidCode.

Поддерживает:
    - Регистрацию интерфейсов и их реализаций (алиасы).
    - Фабрики с опцией shared (синглтон или прототип).
    - Автоматическое разрешение зависимостей по типам в конструкторе.
    - Обнаружение циклических зависимостей.
    - Кэширование разрешённых алиасов.
"""

import inspect
import logging
from typing import (
    Any,
    Callable,
    Dict,
    Optional,
    Set,
    Type,
    TypeVar,
    get_type_hints,
)

T = TypeVar("T")

logger = logging.getLogger(__name__)


class CircularDependencyError(Exception):
    """Исключение, выбрасываемое при обнаружении циклической зависимости."""

    pass


class Container:
    """
    DI-контейнер с автопроводкой (autowiring).

    Поддерживает:
    - Регистрацию интерфейсов и их реализаций (алиасы).
    - Фабрики с опцией shared (синглтон или прототип).
    - Автоматическое разрешение зависимостей по типам в конструкторе.
    - Обнаружение циклических зависимостей.
    - Кэширование разрешённых алиасов.
    """

    def __init__(self) -> None:
        # Хранилище синглтонов: тип -> экземпляр
        self._services: Dict[Type[Any], Any] = {}
        # Фабрики: тип -> (callable, shared_flag)
        self._factories: Dict[Type[Any], tuple] = {}
        # Алиасы: интерфейс -> реализация
        self._aliases: Dict[Type[Any], Type[Any]] = {}
        # Кэш разрешённых алиасов (конечная реализация)
        self._resolved_aliases: Dict[Type[Any], Type[Any]] = {}
        # Множество типов, которые сейчас строятся (для обнаружения циклов)
        self._building: Set[Type[Any]] = set()

    def register(
            self,
            interface: Type[T],
            implementation: Optional[Type[T]] = None,
    ) -> "Container":
        """
        Регистрирует привязку интерфейса к реализации.

        Если implementation не указан, интерфейс считается реализацией.
        Если интерфейс и реализация совпадают, алиас не создаётся.
        """
        if implementation is None:
            implementation = interface
        if implementation != interface:
            self._aliases[interface] = implementation
        return self

    def factory(
            self,
            cls: Type[T],
            factory: Callable[[], T],
            shared: bool = True,
    ) -> "Container":
        """
        Регистрирует фабричную функцию для создания экземпляра.

        Args:
            cls: Тип объекта, который создаёт фабрика.
            factory: Функция, возвращающая экземпляр.
            shared: Если True (по умолчанию) — объект будет синглтоном,
                    если False — каждый вызов get() будет создавать новый.
        """
        self._factories[cls] = (factory, shared)
        return self

    def resolve(self, cls: Type[T]) -> T:
        """Алиас для get()."""
        return self.get(cls)

    def get(self, cls: Type[T]) -> T:
        """
        Возвращает экземпляр сервиса по его типу.

        Если объект уже создан (синглтон) — возвращает его.
        Иначе создаёт новый, разрешая зависимости.
        При обнаружении циклической зависимости выбрасывает исключение.
        """
        key = cls

        # 1. Синглтон
        if key in self._services:
            return self._services[key]

        # 2. Проверка циклической зависимости (при построении)
        if key in self._building:
            raise CircularDependencyError(
                f"Circular dependency detected for {key.__name__}"
            )

        # 3. Разрешение алиасов (с кэшированием)
        if key in self._aliases:
            impl = self._aliases[key]
            if impl not in self._resolved_aliases:
                if impl == key:
                    self._resolved_aliases[key] = key
                else:
                    self._resolved_aliases[key] = self._resolve_alias(impl)
            resolved = self._resolved_aliases[key]
            if resolved != key:
                return self.get(resolved)

        # 4. Фабрика
        if key in self._factories:
            factory, shared = self._factories[key]
            obj = factory()
            if shared:
                self._services[key] = obj
            return obj

        # 5. Создание через конструктор (с автопроводкой)
        self._building.add(key)
        try:
            obj = self._build(key)
        finally:
            self._building.remove(key)

        # Сохраняем как синглтон
        self._services[key] = obj
        return obj

    def _resolve_alias(self, cls: Type[T]) -> Type[T]:
        """
        Рекурсивно разрешает алиас до конечной реализации.

        Обнаруживает циклы в цепочках алиасов.
        """
        current = cls
        visited: Set[Type] = set()
        while current in self._aliases:
            if current in visited:
                raise CircularDependencyError(
                    f"Circular alias chain for {cls.__name__}"
                )
            visited.add(current)
            current = self._aliases[current]
        return current

    def _build(self, cls: Type[T]) -> T:
        """
        Создаёт экземпляр класса, разрешая зависимости конструктора.

        Пропускает параметры *args и **kwargs, так как их невозможно
        разрешить через DI-контейнер.
        """
        # Если конструктор отсутствует или унаследован от object
        if not hasattr(cls, "__init__") or cls.__init__ is object.__init__:
            return cls()

        constructor = cls.__init__
        hints = self._safe_get_type_hints(constructor)
        params = inspect.signature(constructor).parameters

        args = []
        for name, param in params.items():
            if name == "self":
                continue

            # Пропускаем *args и **kwargs
            if param.kind in (
                    inspect.Parameter.VAR_POSITIONAL,
                    inspect.Parameter.VAR_KEYWORD,
            ):
                continue

            # Если есть аннотация типа
            if name in hints:
                param_type = hints[name]
                if inspect.isclass(param_type):
                    try:
                        args.append(self.get(param_type))
                    except CircularDependencyError as e:
                        raise CircularDependencyError(
                            f"While building {cls.__name__}, "
                            f"circular dependency on {param_type.__name__}"
                        ) from e
                else:
                    # Примитивы: используем значение по умолчанию или None
                    if param.default != inspect.Parameter.empty:
                        args.append(param.default)
                    else:
                        args.append(None)
            else:
                # Без аннотации: берём дефолт или None
                if param.default != inspect.Parameter.empty:
                    args.append(param.default)
                else:
                    # Если нет аннотации и нет дефолта – логируем предупреждение
                    logger.warning(
                        f"Parameter '{name}' in {cls.__name__} has no type hint "
                        "and no default value. Will be set to None."
                    )
                    args.append(None)

        return cls(*args)

    def _safe_get_type_hints(self, func: Callable) -> Dict[str, Any]:
        """
        Безопасно получает аннотации типов для функции.

        Если встречается NameError (например, из-за строковой аннотации
        без импорта from __future__ import annotations), выбрасывает
        понятное исключение.
        """
        try:
            return get_type_hints(func)
        except NameError as e:
            raise NameError(
                f"Unable to resolve type hints for {func.__qualname__}. "
                "Ensure all types are defined or import "
                "'from __future__ import annotations'."
            ) from e

    def set(self, cls: Type[T], instance: T) -> "Container":
        """
        Вручную устанавливает готовый объект в контейнер как синглтон.
        """
        self._services[cls] = instance
        return self