src/liquidcode/pydantic.py - develop-0.4 - liquid-g/liquid-code

src/liquidcode/pydantic.py
268 строк9 KB
User
0.4.11 - интеграция Pydantic для автоматической сериализации и десериализации моделей
06 июл 2026, 14:58
06 июл 2026, 14:58d729128
"""
Pydantic-интеграция для LiquidCode.

Предоставляет функции для автоматической сериализации и десериализации
Pydantic-моделей в JSON-совместимые типы.

Опциональная интеграция: если Pydantic не установлен, все функции
возвращают результат без изменений.
"""

from typing import Any, Dict, List, Optional, Type, Union, get_args, get_origin

# Ленивая загрузка для избежания циклических импортов
_pydantic_base_model = None


def _get_base_model() -> Optional[type]:
    """Ленивая загрузка BaseModel из pydantic."""
    global _pydantic_base_model
    if _pydantic_base_model is not None:
        return _pydantic_base_model

    try:
        from pydantic import BaseModel

        # Проверка версии Pydantic
        # Pydantic v2: BaseModel
        # Pydantic v1: BaseModel (но методы другие)
        _pydantic_base_model = BaseModel
        return BaseModel
    except ImportError:
        return None


def _is_pydantic_model_class(annotation: Type) -> bool:
    """Проверяет, является ли тип классом Pydantic BaseModel."""
    if not isinstance(annotation, type):
        return False

    base_model = _get_base_model()
    if base_model is None:
        return False

    try:
        return issubclass(annotation, base_model)
    except TypeError:
        return False


def is_pydantic_model(annotation: Any) -> bool:
    """
    Проверяет, является ли аннотация Pydantic-моделью.
    
    Поддерживает:
        - BaseModel (прямой тип)
        - Optional[BaseModel]
        - List[BaseModel]
        - List[Optional[BaseModel]]
    
    Args:
        annotation: Тип для проверки.
    
    Returns:
        True, если тип — Pydantic модель (или контейнер с моделями).
    """
    if annotation is None or annotation is type(None):
        return False

    # Простой класс
    if _is_pydantic_model_class(annotation):
        return True

    # Проверяем Generic-типы (Optional, List, etc.)
    origin = get_origin(annotation)
    if origin is None:
        return False

    # Получаем аргументы generic-типа
    args = get_args(annotation)

    if origin in (list, List):
        # List[Model] или List[Optional[Model]]
        return len(args) > 0 and (
                _is_pydantic_model_class(args[0]) or
                is_pydantic_model(args[0])
        )

    if origin in (dict, Dict):
        # Dict[str, Model]
        if len(args) >= 2:
            return _is_pydantic_model_class(args[1])
        return False

    if origin in (Union, Optional):
        # Optional[Model] или Union[Model, None]
        return any(is_pydantic_model(arg) for arg in args if arg is not type(None))

    return False


def serialize_response(result: Any) -> Any:
    """
    Сериализует Pydantic-модели в dict/list[dict].
    
    Поддерживает:
        - Одиночную модель → dict
        - Список моделей → list[dict]
        - Tuple/Set из моделей → list[dict]
        - Dict[str, Model] → dict[str, dict]
    
    Args:
        result: Результат контроллера (может быть Pydantic моделью).
    
    Returns:
        Сериализованный результат (dict/list или значение без изменений).
    """
    if result is None:
        return None

    base_model = _get_base_model()
    if base_model is None:
        return result

    # Проверка на одиночную модель
    if isinstance(result, base_model):
        try:
            # Pydantic v2: model_dump()
            if hasattr(result, 'model_dump'):
                return result.model_dump()
            # Pydantic v1: dict()
            elif hasattr(result, 'dict'):
                return result.dict()
        except Exception:
            pass
        return result

    # Проверка на список/кортеж/множество
    if isinstance(result, (list, tuple, set)):
        if len(result) == 0:
            return []

        # Проверяем первый элемент
        first = result[0]
        if isinstance(first, base_model):
            try:
                serialized = []
                for item in result:
                    if isinstance(item, base_model):
                        if hasattr(item, 'model_dump'):
                            serialized.append(item.model_dump())
                        elif hasattr(item, 'dict'):
                            serialized.append(item.dict())
                    else:
                        serialized.append(item)
                return serialized
            except Exception:
                pass
        return list(result)

    # Проверка на dict с Pydantic-моделями в значениях
    if isinstance(result, dict):
        try:
            serialized = {}
            for key, value in result.items():
                if isinstance(value, base_model):
                    if hasattr(value, 'model_dump'):
                        serialized[key] = value.model_dump()
                    elif hasattr(value, 'dict'):
                        serialized[key] = value.dict()
                else:
                    serialized[key] = value
            return serialized
        except Exception:
            pass

    return result


def deserialize_body(body: bytes, annotation: Any) -> Any:
    """
    Десериализует JSON-тело в Pydantic-модель.
    
    Args:
        body: Тело запроса в байтах.
        annotation: Ожидаемый тип (должен быть Pydantic-моделью).
    
    Returns:
        Десериализованная модель или None.
    
    Raises:
        ValidationError: Если валидация не удалась.
    """
    if body is None or len(body) == 0:
        return None

    base_model = _get_base_model()
    if base_model is None:
        # Если Pydantic не установлен, просто возвращаем JSON
        try:
            import json
            return json.loads(body.decode('utf-8'))
        except Exception:
            return None

    # Проверяем, является ли annotation моделью
    if not is_pydantic_model(annotation):
        try:
            import json
            return json.loads(body.decode('utf-8'))
        except Exception:
            return None

    try:
        # Извлекаем базовый тип (убираем Optional, List и т.д.)
        origin = get_origin(annotation)
        target_annotation = annotation

        if origin in (list, List):
            # List[Model] - берем первый аргумент
            args = get_args(annotation)
            if args:
                target_annotation = args[0]
        elif origin in (Union, Optional):
            # Optional[Model] - берем ненулевой аргумент
            args = get_args(annotation)
            for arg in args:
                if arg is not type(None) and _is_pydantic_model_class(arg):
                    target_annotation = arg
                    break
        elif _is_pydantic_model_class(annotation):
            # Простая модель
            target_annotation = annotation
        else:
            # Не модель
            try:
                import json
                return json.loads(body.decode('utf-8'))
            except Exception:
                return None

        # Десериализация JSON → dict → модель
        import json
        data = json.loads(body.decode('utf-8'))

        # Если это список моделей
        if origin in (list, List):
            result = []
            if isinstance(data, list):
                for item in data:
                    if isinstance(item, dict):
                        result.append(target_annotation.model_validate(item))
                    else:
                        result.append(item)
            else:
                # Single item в списке
                if isinstance(data, dict):
                    result.append(target_annotation.model_validate(data))
            return result

        # Одиночная модель
        if isinstance(data, dict):
            return target_annotation.model_validate(data)

        return data

    except Exception as e:
        from .validation import ValidationError
        raise ValidationError({"body": [f"Validation error: {str(e)}"]})