Профилирование Pydantic, кэширование, lazy-валидация, сравнение v1 и v2.
Pydantic v2 значительно быстрее v1 благодаря Rust-ядру. Изучим, как выжать максимум производительности
Pydantic v2 использует Rust-библиотеку pydantic-core:
# Pydantic v1: ~100% (базовая линия)
# Pydantic v2: ~5-50x быстрее в зависимости от сценария
from pydantic import BaseModel, Field
import timeit
class User(BaseModel):
id: int
name: str
email: str
age: int = Field(ge=0, le=150)
# Тест: 10000 валидаций
def test_v2():
for i in range(10000):
User(id=i, name="Test", email="test@example.com", age=25)
# Время: ~0.05 секунды (v2)
# Время: ~0.5 секунды (v1)import cProfile
import pstats
from pydantic import BaseModel, Field
class ComplexModel(BaseModel):
field1: str = Field(min_length=1, max_length=100)
field2: int = Field(ge=0, le=1000)
field3: list[str] = Field(min_length=1)
field4: dict[str, int]
def validate_many():
for i in range(1000):
ComplexModel(
field1="test",
field2=500,
field3=["a", "b", "c"],
field4={"key": 1}
)
# Профилирование
profiler = cProfile.Profile()
profiler.enable()
validate_many()
profiler.disable()
# Статистика
stats = pstats.Stats(profiler)
stats.sort_stats('cumulative')
stats.print_stats(10)from functools import lru_cache
from pydantic import BaseModel, TypeAdapter
# TypeAdapter для переиспользования
user_adapter = TypeAdapter(dict[str, str | int])
@lru_cache(maxsize=1000)
def validate_user_cached(data_tuple: tuple) -> dict:
"""Кэширование валидации для неизменных данных"""
data_dict = dict(zip(data_dict[::2], data_dict[1::2]))
return user_adapter.validate_python(data_dict)
# Для изменяемых данных используйте хэш
from hashlib import md5
validation_cache = {}
def validate_with_cache(model_class, data: dict):
"""Кэширование с хэшированием"""
data_hash = md5(str(sorted(data.items())).encode()).hexdigest()
if data_hash in validation_cache:
return validation_cache[data_hash]
result = model_class(**data)
validation_cache[data_hash] = result
if len(validation_cache) > 1000:
# Очистка старых записей
validation_cache.clear()
return resultfrom pydantic import BaseModel, model_validator
from typing import Any
class LazyModel(BaseModel):
data: dict[str, Any]
_validated: bool = False
@model_validator(mode='after')
def lazy_validate(self) -> 'LazyModel':
# Валидация откладывается до первого доступа
self._validated = False
return self
def __getattribute__(self, name: str):
# Валидация при первом доступе к data
if name == 'data' and not object.__getattribute__(self, '_validated'):
data = object.__getattribute__(self, 'data')
# Валидация данных
object.__setattr__(self, '_validated', True)
return object.__getattribute__(self, name)from pydantic import TypeAdapter
from typing import List
# НЕЭФФЕКТИВНО: валидация по одному
def validate_items_one_by_one(items: list[dict]) -> list:
adapter = TypeAdapter(dict[str, int])
return [adapter.validate_python(item) for item in items]
# ЭФФЕКТИВНО: валидация списком
def validate_items_batch(items: list[dict]) -> list:
adapter = TypeAdapter(list[dict[str, int]])
return adapter.validate_python(items)
# Разница: до 10x быстрее для больших списковfrom pydantic import BaseModel, ConfigDict
class FastModel(BaseModel):
model_config = ConfigDict(
validate_assignment=False, # отключить валидацию при присваивании
extra='ignore' # не проверять лишние поля
)
field1: str
field2: int
# Быстрое создание без полной валидации
instance = FastModel.model_construct(
field1="value",
field2=42
)
# model_construct пропускает валидацию для производительностиfrom pydantic import BaseModel, ConfigDict
class SlottedModel(BaseModel):
model_config = ConfigDict(
extra='forbid',
frozen=True # требуется для __slots__
)
__slots__ = ('field1', 'field2')
field1: str
field2: int
# Меньше памяти, быстрее доступ
# Но нельзя добавлять атрибуты динамическиfrom pydantic import BaseModel
import json
class Model(BaseModel):
field1: str
field2: int
field3: list[str]
# Предварительная генерация JSON Schema
schema = Model.model_json_schema()
schema_json = json.dumps(schema)
# Использование схемы для валидации JSON
from pydantic import TypeAdapter
adapter = TypeAdapter(Model)
data = adapter.validate_json(json_string)from pydantic import BaseModel
class User(BaseModel):
name: str
age: int
data = {"name": "Alice", "age": 25}
# Конструктор (быстрее для dict)
user1 = User(**data)
# model_validate (универсальнее)
user2 = User.model_validate(data)
# model_validate_json (для JSON-строк)
user3 = User.model_validate_json('{"name": "Alice", "age": 25}')
# Производительность:
# **data: ~100% (базовая)
# model_validate: ~110-120%
# model_validate_json: ~90% (быстрее для JSON)import asyncio
from pydantic import BaseModel
from concurrent.futures import ThreadPoolExecutor
class Data(BaseModel):
value: int
async def validate_async(data_list: list[dict]) -> list[Data]:
"""Асинхронная валидация списка"""
loop = asyncio.get_event_loop()
with ThreadPoolExecutor() as executor:
# Параллельная валидация
futures = [
loop.run_in_executor(executor, Data, **data)
for data in data_list
]
results = await asyncio.gather(*futures)
return list(results)
# Для CPU-bound валидации используйте ProcessPoolExecutorimport timeit
from pydantic import BaseModel, Field, TypeAdapter
class Model(BaseModel):
field1: str = Field(min_length=1)
field2: int = Field(ge=0)
field3: list[str]
# Тест 1: Конструктор
def test_constructor():
return Model(field1="test", field2=42, field3=["a", "b"])
# Тест 2: model_validate
def test_model_validate():
return Model.model_validate({"field1": "test", "field2": 42, "field3": ["a", "b"]})
# Тест 3: TypeAdapter
adapter = TypeAdapter(Model)
def test_type_adapter():
return adapter.validate_python({"field1": "test", "field2": 42, "field3": ["a", "b"]})
# Тест 4: model_construct (без валидации)
def test_model_construct():
return Model.model_construct(field1="test", field2=42, field3=["a", "b"])
# Результаты (10000 итераций):
# constructor: ~0.01s
# model_validate: ~0.012s
# type_adapter: ~0.012s
# model_construct: ~0.001s (но без валидации!)Создайте высокопроизводительный валидатор для массовых данных:
from pydantic import BaseModel, TypeAdapter, ValidationError
from typing import List, Tuple, Optional
from functools import lru_cache
import hashlib
class BatchValidator:
"""Высокопроизводительный валидатор для пакетной обработки"""
def __init__(self, model_class: type[BaseModel]):
self.model_class = model_class
self.adapter = TypeAdapter(List[model_class])
self._cache = {}
def validate_batch(
self,
items: List[dict],
use_cache: bool = True
) -> Tuple[List[BaseModel], List[Tuple[int, str]]]:
"""
Валидация пачки данных.
Возвращает (валидные_элементы, ошибки[(индекс, сообщение)])
"""
valid = []
errors = []
# Группировка одинаковых данных для кэширования
if use_cache:
# Реализуйте кэширование
pass
# Пакетная валидация
for i, item in enumerate(items):
try:
validated = self.model_class(**item)
valid.append(validated)
except ValidationError as e:
errors.append((i, str(e.error_count())))
return valid, errors
def validate_batch_fast(
self,
items: List[dict]
) -> List[BaseModel]:
"""
Быстрая валидация без обработки ошибок.
Подходит для доверенных данных.
"""
return self.adapter.validate_python(items)
def clear_cache(self):
"""Очистка кэша"""
self._cache.clear()
# Пример использования:
class User(BaseModel):
name: str
email: str
age: int
validator = BatchValidator(User)
items = [
{"name": "Alice", "email": "alice@example.com", "age": 25},
{"name": "Bob", "email": "bob@example.com", "age": 30},
# ... 10000 элементов
]
valid, errors = validator.validate_batch(items)
print(f"Валидно: {len(valid)}, Ошибки: {len(errors)}")Далее: Production-паттерны: лучшие практики