src/liquidcode/kernel/argument_resolver.py - master - liquid-g/liquid-code

src/liquidcode/kernel/argument_resolver.py
161 строка6 KB
User
0.3.0-RC - Добавлена система валидации (Validator, базовые валидаторы, интеграция с ArgumentResolver, генератор `make:validator`)
04 июл 2026, 13:30
04 июл 2026, 13:306b33e13
"""
ArgumentResolver — автоматическое преобразование параметров контроллера.

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

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

from ..container import Container
from ..contracts import Request


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],
            body: bytes,
            method: str,
    ) -> Dict[str, Any]:
        """
        Возвращает словарь аргументов для вызова контроллера.

        Args:
            controller_method: Метод контроллера.
            request: Объект запроса.
            route_params: Параметры из маршрута.
            query_params: Параметры из query-строки.
            body: Тело запроса (байты).
            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 = {}

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

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

            # 1. Проверяем, не является ли параметр объектом Request
            # Если аннотация пустая, но имя 'request', считаем это 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 body:
                    try:
                        body_data = json.loads(body.decode('utf-8'))
                        kwargs[name] = body_data
                    except json.JSONDecodeError:
                        kwargs[name] = None
                # Если метод не подразумевает тело, не передаём data (оставляем дефолт или None)
                # Если мы не установили kwargs[name], то пойдём дальше к дефолту
                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