src/liquidcode/kernel/argument_resolver.py - develop-0.4 - liquid-g/liquid-code

src/liquidcode/kernel/argument_resolver.py
168 строк7 KB
User
0.4.18 - fix: устранение дублирования парсинга тела запроса (T-001)
07 июл 2026, 22:07
07 июл 2026, 22:073182883
"""
ArgumentResolver — автоматическое преобразование параметров контроллера.

Поддерживает:
- Параметры маршрута (по имени)
- Query-параметры (по имени)
- Тело запроса (по аннотации `data`)
- Pydantic модели (автоматическая десериализация)
- Объект Request
- Сервисы из DI-контейнера (по типу)
- Приведение типов (int, float, bool, str)
- Значения по умолчанию
"""

import inspect
from typing import Any, Callable, Dict, Optional, Type, get_type_hints

from ..container import Container
from ..contracts import Request
from ..pydantic import is_pydantic_model


class ArgumentResolver:
    """
    Разрешает аргументы для контроллера на основе запроса и контейнера.
    """

    def __init__(self, container: Container):
        self.container = container

    def resolve(
            self,
            controller_method: Callable,
            request: Request,
            route_params: Dict[str, str],
            query_params: Dict[str, str],
            method: str,
    ) -> Dict[str, Any]:
        """
        Возвращает словарь аргументов для вызова контроллера.

        Args:
            controller_method: Метод контроллера.
            request: Объект запроса.
            route_params: Параметры из маршрута.
            query_params: Параметры из query-строки.
            method: HTTP-метод (GET, POST, etc.)

        Returns:
            Словарь {имя_параметра: значение} для передачи в контроллер.
        """
        # Получаем аннотации типов с учётом from __future__ import annotations
        try:
            hints = get_type_hints(controller_method)
        except (NameError, TypeError):
            hints = {}

        sig = inspect.signature(controller_method)
        kwargs = {}

        # Читаем распарсенный body из атрибута запроса (один раз)
        parsed_body = request.get_attr('parsed_body')  # dict или None

        for name, param in sig.parameters.items():
            if name == 'self':
                continue

            annotation = hints.get(name, param.annotation)

            # 1. Проверяем, не является ли параметр объектом Request
            if name == 'request':
                kwargs[name] = request
                continue
            if self._is_request_type(annotation):
                kwargs[name] = request
                continue

            # 2. Параметр из маршрута (приоритет выше query)
            if name in route_params:
                kwargs[name] = self._cast_value(route_params[name], annotation)
                continue

            # 3. Параметр из query
            if name in query_params:
                kwargs[name] = self._cast_value(query_params[name], annotation)
                continue

            # 4. Специальный параметр 'data' для тела запроса (POST/PUT/PATCH)
            if name == 'data':
                if method.upper() in ('POST', 'PUT', 'PATCH') and parsed_body is not None:
                    # Если annotation — Pydantic модель, десериализуем из dict (v2)
                    if is_pydantic_model(annotation):
                        # Pydantic v2: model_validate
                        if hasattr(annotation, 'model_validate'):
                            kwargs[name] = annotation.model_validate(parsed_body)
                        # fallback для v1 (маловероятно, но оставим)
                        elif hasattr(annotation, 'parse_obj'):
                            kwargs[name] = annotation.parse_obj(parsed_body)
                    else:
                        # Просто отдаём распарсенный dict
                        kwargs[name] = parsed_body
                # Если метод не подразумевает тело или parsed_body None, ничего не делаем
                # (пойдём к дефолту или None)
                if name in kwargs:
                    continue

            # 5. Параметр, который нужно получить из контейнера (сервис)
            if self._is_service_type(annotation):
                try:
                    kwargs[name] = self.container.get(annotation)
                    continue
                except Exception:
                    # Если сервис не найден, пробуем дефолтное значение
                    pass

            # 6. Если есть значение по умолчанию
            if param.default != inspect.Parameter.empty:
                kwargs[name] = param.default
                continue

            # 7. Иначе None
            kwargs[name] = None

        return kwargs

    def _cast_value(self, value: str, annotation: Optional[Type]) -> Any:
        """Приводит строковое значение к нужному типу на основе аннотации."""
        if annotation == inspect.Parameter.empty or annotation is None:
            return value
        if not isinstance(annotation, type):
            return value
        if annotation is int:
            return int(value)
        if annotation is float:
            return float(value)
        if annotation is bool:
            return value.lower() in ('true', '1', 'yes', 'on')
        return value

    def _is_request_type(self, annotation) -> bool:
        """Проверяет, является ли аннотация типом Request."""
        if annotation == inspect.Parameter.empty:
            return False
        if not isinstance(annotation, type):
            return False
        try:
            return issubclass(annotation, Request)
        except TypeError:
            return False

    def _is_service_type(self, annotation) -> bool:
        """
        Проверяет, является ли аннотация классом сервиса,
        зарегистрированного в контейнере (исключая примитивные типы).
        """
        if annotation == inspect.Parameter.empty or annotation is None:
            return False
        if not isinstance(annotation, type):
            return False
        # Исключаем примитивные типы (они не сервисы)
        if annotation in (str, int, float, bool, dict, list, tuple, type(None)):
            return False
        # Проверяем, зарегистрирован ли класс в контейнере
        try:
            self.container.get(annotation)
            return True
        except Exception:
            return False