140 lines
5.2 KiB
Python
140 lines
5.2 KiB
Python
"""
|
||
Упрощенные утилиты для работы со схемами Pydantic
|
||
"""
|
||
from typing import List, Dict, Any, Type
|
||
from pydantic import BaseModel
|
||
import inspect
|
||
|
||
|
||
def get_required_fields_from_schema(schema_class: Type[BaseModel]) -> List[str]:
|
||
"""
|
||
Извлекает список обязательных полей из схемы Pydantic
|
||
|
||
Args:
|
||
schema_class: Класс схемы Pydantic
|
||
|
||
Returns:
|
||
Список имен обязательных полей
|
||
"""
|
||
required_fields = []
|
||
|
||
# Используем model_fields для Pydantic v2 или __fields__ для v1
|
||
if hasattr(schema_class, 'model_fields'):
|
||
fields = schema_class.model_fields
|
||
else:
|
||
fields = schema_class.__fields__
|
||
|
||
for field_name, field_info in fields.items():
|
||
# В Pydantic v2 есть метод is_required()
|
||
if hasattr(field_info, 'is_required'):
|
||
if field_info.is_required():
|
||
required_fields.append(field_name)
|
||
elif hasattr(field_info, 'required'):
|
||
if field_info.required:
|
||
required_fields.append(field_name)
|
||
else:
|
||
# Fallback для старых версий - проверяем наличие default
|
||
has_default = False
|
||
|
||
if hasattr(field_info, 'default'):
|
||
has_default = field_info.default is not ...
|
||
elif hasattr(field_info, 'default_factory'):
|
||
has_default = field_info.default_factory is not None
|
||
|
||
if not has_default:
|
||
required_fields.append(field_name)
|
||
|
||
return required_fields
|
||
|
||
|
||
def get_optional_fields_from_schema(schema_class: Type[BaseModel]) -> List[str]:
|
||
"""
|
||
Извлекает список необязательных полей из схемы Pydantic
|
||
|
||
Args:
|
||
schema_class: Класс схемы Pydantic
|
||
|
||
Returns:
|
||
Список имен необязательных полей
|
||
"""
|
||
optional_fields = []
|
||
|
||
# Используем model_fields для Pydantic v2 или __fields__ для v1
|
||
if hasattr(schema_class, 'model_fields'):
|
||
fields = schema_class.model_fields
|
||
else:
|
||
fields = schema_class.__fields__
|
||
|
||
for field_name, field_info in fields.items():
|
||
# В Pydantic v2 есть метод is_required()
|
||
if hasattr(field_info, 'is_required'):
|
||
if not field_info.is_required():
|
||
optional_fields.append(field_name)
|
||
elif hasattr(field_info, 'required'):
|
||
if not field_info.required:
|
||
optional_fields.append(field_name)
|
||
else:
|
||
# Fallback для старых версий - проверяем наличие default
|
||
has_default = False
|
||
|
||
if hasattr(field_info, 'default'):
|
||
has_default = field_info.default is not ...
|
||
elif hasattr(field_info, 'default_factory'):
|
||
has_default = field_info.default_factory is not None
|
||
|
||
if has_default:
|
||
optional_fields.append(field_name)
|
||
|
||
return optional_fields
|
||
|
||
|
||
def register_getter_from_schema(parser_instance, getter_name: str, method: callable,
|
||
schema_class: Type[BaseModel], description: str = ""):
|
||
"""
|
||
Регистрирует геттер в парсере, используя схему Pydantic для определения параметров
|
||
|
||
Args:
|
||
parser_instance: Экземпляр парсера
|
||
getter_name: Имя геттера
|
||
method: Метод для выполнения
|
||
schema_class: Класс схемы Pydantic
|
||
description: Описание геттера (если не указано, берется из docstring метода)
|
||
"""
|
||
# Извлекаем параметры из схемы
|
||
required_params = get_required_fields_from_schema(schema_class)
|
||
optional_params = get_optional_fields_from_schema(schema_class)
|
||
|
||
# Если описание не указано, берем из docstring метода
|
||
if not description:
|
||
description = inspect.getdoc(method) or ""
|
||
|
||
# Регистрируем геттер
|
||
parser_instance.register_getter(
|
||
name=getter_name,
|
||
method=method,
|
||
required_params=required_params,
|
||
optional_params=optional_params,
|
||
description=description
|
||
)
|
||
|
||
|
||
def validate_params_with_schema(params: Dict[str, Any], schema_class: Type[BaseModel]) -> Dict[str, Any]:
|
||
"""
|
||
Валидирует параметры с помощью схемы Pydantic
|
||
|
||
Args:
|
||
params: Словарь параметров
|
||
schema_class: Класс схемы Pydantic
|
||
|
||
Returns:
|
||
Валидированные параметры
|
||
|
||
Raises:
|
||
ValidationError: Если параметры не прошли валидацию
|
||
"""
|
||
try:
|
||
# Создаем экземпляр схемы для валидации
|
||
validated_data = schema_class(**params)
|
||
return validated_data.dict()
|
||
except Exception as e:
|
||
raise ValueError(f"Ошибка валидации параметров: {str(e)}") |