Давайте ещё раз взглянем на этот код (теперь уж с нормальной подсветкой):

def handler_create_user(r: Request):
    input_data = r.post()
    first_name, email = input_data.get('firstName'), input_data.get('email')
    if not first_name or not email:
        raise HTTPBadRequest('name & email must have values')
    return User.create(name=first_name, email=email, password=uuid4())

Итак, какие тут есть узкие места?

  • Мы понятия не имеем, что будет содержать тело запроса (какие именно ключи), поэтому достаем, надеясь на успех, через get, а потом вынуждены проверять, что досталось всё требуемое

  • Конвертация стилей наименований переменных (Snake case, Camel case, etc.) тоже ложится на хрупкие плечи программиста

  • Приходится самому писать получение всех значений из тела запроса, т.о. множество строк вида my_var = input_data.get('my_var') могут значительно “загрязнить” код

  • Валидация значений происходит тут же (в хендлере), и она смешивается с бизнес-логикой

  • Формат ответа не определен явно и зависит от внешних факторов (в данном случае – от модели в базе данных). Т.о. при добавлении нового поля в модель, оно автоматически возникнет и в ответе, что может сломать API

  • В response могут попасть нежелательные поля (например, пароль или секретный ключ)

Можно ли этот код сделать хуже? Конечно!

def handler_create_user(r: Request):
    return User.create(**r.post(), password=uuid4())

А вот что требуется, чтобы сделать лучше, давайте разбираться.

Как это готовить

Сразу оговорюсь, что будем рассматривать валидацию данных в разрезе работы API бекенда веб-сервиса, для desktop систем подход может оказаться иным.

  • Есть случаи, когда валидация не нужна. Например, вы стартап, срок жизни кода – одна неделя, некогда делать хорошо, надо делать быстро

  • Валидация – это контракт, проверять который стоит лишь на стыке систем, потому что неизвестно, что именно пришло извне, а далее, внутри системы, мы сами контролируем данные. Можно об этом думать как о некой границе, которую данные проходят, а далее никто паспорт проверять не будет (да и не надо, ведь все преобразования уже под нашим контролем)

  • Причем, эту границу данные проходят как по “приезде”, так и при “отъезде”. Важно контролировать не только вход, но и выход, ибо может оказаться, что не все поля строки из ORM модели вы хотите показывать наружу, да и самой модели может не быть, т.к. ответ формировался из разных источников. Кроме того, это даст контракт пользователю вашего API: чёткую структуру ответа, на которую можно положиться

  • Валидацию полезно совмещать с конвертацией (приведением, cast) данных. Например, если мы ждём в строковом поле дату, то будет приятно, если после проверки, что значение поля действительно дата, произойдет еще конвертация str -> datetime; или строка превратится в enum

  • Если мы делаем условный сайт или мобильное приложение, то проверки должны быть на двух уровнях: на стороне фронта (защита от дурака) и на стороне бека (защита от хакера). Порой первым пренебрегают в угоду скорости разработки (получая бонусом ухудшение UX и увеличение нагрузки на сеть; привет, Хабр), но второй – это последний рубеж обороны, иначе в будущем вас будут ждать миграции данных в базе, а вы их точно не хотите делать

  • В MVC-like паттернах представление (view) будет более чистым, т.к. вся валидация инкапсулирована в схемах. Да и тестировать по частям удобнее

Show me the code

Сравним четыре Python библиотеки для валидации данных на синтетическом примере, чтобы показать использование наиболее частых типов данных для перекладывателя json’ов (a.k.a. backend developer) а так же написание кастомных проверок. Бонусом получим некую шпаргалку, которая может быть полезна для того, чтобы освежить память при переключении между проектами с разными библиотеками.

Будем подавать на вход словарь, чтобы не зависеть от веб-фреймворка, т.е. не рассматриваем, как именно из запроса (request) получили этот словарь и как потом из словаря создастся json для ответа (response).

Отдельно обращу внимание, что иногда хочется с помощью механизма валидации ещё и проверить данные во внешних источниках. Например, одно из полей запроса является ключом в одной из таблиц базы данных, поэтому может появиться желание в рамках валидации еще выполнить sql запрос. Это похвально, но прокидывание зависимостей (dependency) в схемы валидации откроет портал в ад, поэтому лучше подобное делать на более поздних этапах в специально отведённых местах.

Pydantic

Довольно популярная библиотека в последнее время, т.к. является частью хайпящего FastApi.

Код схемы
from typing import Annotated, ClassVar, Literal

from pydantic import (
    UUID4,
    BaseModel,
    ConfigDict,
    EmailStr,
    Field,
    HttpUrl,
    NonNegativeInt,
    PastDatetime,
    PositiveInt,
    field_validator,
    model_validator,
)
from pydantic_extra_types.country import CountryAlpha3
from pydantic_extra_types.payment import PaymentCardNumber

from validators.common import Gender


class DocumentSchema(BaseModel):
    number: Annotated[int | str, Field(alias='full_number')]


class PydanticSchema(BaseModel):
    model_config: ClassVar = ConfigDict(extra='ignore')
    
    schema_version: Literal['3.14.15']
    id: UUID4
    created_at: PastDatetime
    name: Annotated[str, Field(min_length=2, max_length=32)]
    age: Annotated[int, Field(ge=0, le=100)]
    is_client: bool
    gender: Gender
    email: EmailStr
    social_profile_url: HttpUrl
    bank_cards: list[PaymentCardNumber] | None
    countries: Annotated[str, Field(min_length=1, max_length=64)]
    document: DocumentSchema
    page_number: NonNegativeInt
    page_size: PositiveInt

    @field_validator('age', mode='after')
    @classmethod
    def check_adults(cls, value: int) -> int:
        if value < 18:
            raise ValueError('only adults')
        return value

    @field_validator('countries')
    @classmethod
    def parse_counties(cls, value: str) -> list[CountryAlpha3]:
        return [CountryAlpha3(c) for c in value.split(',')]

    @model_validator(mode='before')
    @classmethod
    def general_check(cls, data: dict) -> dict:
        if data.get('is_client') and not data.get('bank_cards'):
            raise ValueError('cards are required for clients')
        return data

Тут в хвост и гриву используется стандартный механизм типов, что делает код довольно лаконичным и приятным. Некоторые дополнительные типы раньше были частью Pydantic, но сейчас вынесены в отдельную библиотеку.

Marshmallow

В моей практике обычно используется в связке с aiohttp или Flask.

Код схемы
import typing
from datetime import datetime

from marshmallow import (
    EXCLUDE,
    Schema,
    ValidationError,
    fields,
    validate,
    validates,
    validates_schema,
)
from marshmallow.utils import missing as missing_
from marshmallow_union import Union as UnionField

from validators.common import Gender


class CommaList(fields.Field):
    def __init__(
        self,
        *,
        load_default: typing.Any = missing_,
        missing: typing.Any = missing_,
        dump_default: typing.Any = missing_,
        default: typing.Any = missing_,
        data_key: str | None = None,
        attribute: str | None = None,
        validate: (
            None
            | typing.Callable[[typing.Any], typing.Any]
            | typing.Iterable[typing.Callable[[typing.Any], typing.Any]]
        ) = None,
        required: bool = False,
        allow_none: bool | None = None,
        load_only: bool = False,
        dump_only: bool = False,
        error_messages: dict[str, str] | None = None,
        metadata: typing.Mapping[str, typing.Any] | None = None,
        **additional_metadata,
    ) -> None:
        super().__init__(
            load_default=load_default,
            missing=missing,
            dump_default=dump_default,
            default=default,
            data_key=data_key,
            attribute=attribute,
            validate=validate,
            required=required,
            allow_none=allow_none,
            load_only=load_only,
            dump_only=dump_only,
            error_messages=error_messages,
            metadata=metadata,
            **additional_metadata,
        )
        marshmallow_type = metadata.get('marshmallow_type') if metadata else None
        self.marshmallow_type = marshmallow_type or (lambda x: x)

    def _deserialize(self, value, attr, data, **kwargs) -> list:
        try:
            return [self.marshmallow_type(x) for x in value.split(',')]
        except (ValueError, AttributeError, TypeError) as exc:
            raise ValidationError('Incorrect list') from exc


class DocumentSchema(Schema):
    class Meta:
        unknown = EXCLUDE

    number = UnionField(
        data_key='full_number',
        fields=[fields.Integer(), fields.Str()],
        required=True,
    )


class MarshmallowSchema(Schema):
    class Meta:
        unknown = EXCLUDE

    schema_version = fields.Str(required=True, validate=validate.Equal('3.14.15'))
    id = fields.UUID(required=True)
    created_at = fields.DateTime(required=True)
    name = fields.Str(required=True, validate=validate.Length(min=2, max=32))
    age = fields.Int(required=True, validate=validate.Range(min=0, max=100))
    is_client = fields.Bool(required=True)
    gender = fields.Enum(Gender, by_value=True, required=True)
    email = fields.Email(required=True)
    social_profile_url = fields.URL(required=True)
    bank_cards = fields.List(
        fields.Str(validate=validate.Length(min=15)),
        required=True,
        validate=validate.Length(min=1),
    )
    countries = CommaList(required=True, metadata={'marshmallow_type': str})
    document = fields.Nested(DocumentSchema)
    page_number = fields.Int(required=True, validate=validate.Range(min=0, max=100))
    page_size = fields.Int(required=True, validate=validate.Range(min=1, max=100))

    @validates('created_at')
    def date_must_be_in_past(self, value: datetime) -> None:
        if value >= datetime.utcnow():
            raise ValidationError('date must be in the past')

    @validates('age')
    def check_adults(self, value: int) -> None:
        if value < 18:
            raise ValidationError('only adults')

    @validates_schema
    def general_check(self, data: dict, **kwargs) -> None:
        if data.get('is_client') and not data.get('bank_cards'):
            raise ValidationError('cards are required for clients')

Тут используется иная концепция: "присваивание" вместо типов, – из-за этого получается весьма многословно. Кроме того, пришлось реализовывать CommaList самостоятельно.

Trafaret

Очень редкий зверь с самым экзотическим синтаксисом.

Код схемы
import uuid
from datetime import datetime

import trafaret as t

from validators.common import Gender


class UUID(t.Trafaret):
    def check_and_return(self, value: uuid.UUID | bytes | str | None) -> uuid.UUID | None:
        if value is None:
            return None
        if isinstance(value, uuid.UUID):
            return value
        try:
            if isinstance(value, bytes) and len(value) == 16:
                return uuid.UUID(bytes=value)
            else:
                return uuid.UUID(value)
        except (ValueError, AttributeError, TypeError):
            self._failure('value is not a uuid')


class CommaList(t.Trafaret):
    def __init__(self, *args, trafaret_type: t.Trafaret, **kwargs) -> None:
        super().__init__(*args, **kwargs)
        self.trafaret_type = trafaret_type

    def check_and_return(self, data: str) -> list:
        return [self.trafaret_type.check_and_return(x) for x in data.split(',')]


def past_date(frmt: str = '%Y-%m-%dT%H:%M:%S') -> t.And:
    def check(value: str) -> datetime:
        converted_value = datetime.fromisoformat(value)
        if converted_value >= datetime.utcnow():
            raise t.DataError('date must be in the past')
        return converted_value

    return t.DateTime(format=frmt) >> check


def check_adults(value: int) -> int:
    if value < 18:
        raise t.DataError('only adults')
    return value


def check_schema(data: dict) -> dict:
    if data.get('is_client') and not data.get('bank_cards'):
        raise t.DataError('cards are required for clients')
    return data


document_schema = t.Dict(
    {
        t.Key('full_number') >> 'number': t.Int() | t.String(),
    },
)

trafaret_schema = (
    t.Dict(
        {
            'schema_version': t.Atom('3.14.15'),
            'id': UUID,
            'created_at': past_date(),
            'name': t.String(min_length=2, max_length=32),
            'age': t.Int(gte=0, lte=100) >> check_adults,
            'is_client': t.Bool(),
            'gender': t.Enum(*[i.value for i in Gender]),
            'email': t.Email,
            'social_profile_url': t.URL,
            'bank_cards': t.List(t.Int(gte=10**15), min_length=1),
            'countries': CommaList(trafaret_type=t.String()),
            'document': document_schema,
            'page_number': t.Int(gte=0, lte=100),
            'page_size': t.Int(gte=1, lte=100),
        },
        ignore_extra='*',
    )
    >> check_schema
)

Вновь вернулись к типам, пусть и нестандартным. Полученная схема выглядит аскетично, но типов "из коробки" в этой библиотеке, пожалуй, меньше всех, поэтому придётся их реализовывать самостоятельно.

После двух предыдущих библиотек синтаксис Trafaret может немного отпугнуть, но через некоторое время использования понимаешь, что в этом что-то есть.

Django REST framework (DRF)

Сложно отделить DRF от Django, как можно это сделать в случае Pydantic и FastApi. Поэтому нельзя просто так взять и валидировать словарь, нужно бороться с волочащимся Django: например, делать monkeypatch для настроек (django.settings). В этом разрезе самая проблемная библиотека из обозреваемых.

Код схемы
from datetime import datetime

from rest_framework import serializers as s

from validators.common import Gender


def check_schema_version(value: str) -> None:
    if value != '3.14.15':
        raise s.ValidationError('value must be equal "3.14.15"')


class CommaStrListField(s.Field):
    def to_representation(self, value: list[str]) -> list[str]:
        return value

    def to_internal_value(self, data: str) -> list[str]:
        try:
            return [str(x) for x in data.split(',')]
        except (ValueError, AttributeError, TypeError) as exc:
            raise s.ValidationError('Incorrect list') from exc


class IntOrStrField(s.Field):
    def to_representation(self, value: int | str) -> int | str:
        return value

    def to_internal_value(self, data: int | str) -> int | str:
        return data


class DocumentSchema(s.Serializer):
    full_number = IntOrStrField(source='number')


class DRFSchema(s.Serializer):
    class Meta:
        unknown = 'ignore'

    schema_version = s.CharField(validators=[check_schema_version])
    id = s.UUIDField()
    created_at = s.DateTimeField()
    name = s.CharField(min_length=2, max_length=32)
    age = s.IntegerField(min_value=0, max_value=100)
    is_client = s.BooleanField()
    gender = s.ChoiceField(choices=[(e.value, e.name) for e in Gender])
    email = s.EmailField()
    social_profile_url = s.URLField()
    bank_cards = s.ListField(child=s.CharField(min_length=15))
    countries = CommaStrListField()
    document = DocumentSchema()
    page_number = s.IntegerField(min_value=0, max_value=100)
    page_size = s.IntegerField(min_value=1, max_value=100)

    def validate_created_at(self, value: datetime) -> datetime:
        if value >= datetime.utcnow():
            raise s.ValidationError('date must be in the past')
        return value

    def validate_age(self, value: int) -> int:
        if value < 18:
            raise s.ValidationError('only adults')
        return value

    def validate(self, data: dict) -> dict:
        if data.get('is_client') and not data.get('bank_cards'):
            raise s.ValidationError('cards are required for clients')
        return data

И вновь ушли от типов. Не избежали создания кастомных валидаторов, которые могли бы быть "из коробки".

DRF принуждает нас использовать «единый» стиль для названий кастомных валидаторов. Не уверен, что это можно однозначно отнести к плюсам или минусам.

Также странно, что для каждого вида Union приходится писать свой тип. Возможно, тут что-то не так делаю, в противном случае выглядит как спорное решение.

Тестирование производительности

У каждой библиотеки в закромах есть свой тест, который показывает, что именно эта либа наиболее крутая, я же напишу свой. Он будет весьма неточный, но прикинуть палец к носу с его помощью можно.

Формирование тестовых данных
from enum import StrEnum, unique

import pytest
from faker import Faker


@pytest.fixture(scope='session')
def faker() -> Faker:
    return Faker('en_GB')


@unique
class Gender(StrEnum):
    MALE = 'male'
    FEMALE = 'female'
    HELICOPTER = 'helicopter'


@pytest.fixture
def data(faker: Faker) -> dict:
    return {
        'schema_version': '3.14.15',
        'id': faker.uuid4(cast_to=None),
        'created_at': faker.past_datetime().isoformat().split('.')[0],
        'name': faker.name(),
        'age': faker.pyint(min_value=18, max_value=100),
        'is_client': faker.pybool(),
        'gender': faker.enum(Gender).value,
        'email': faker.email(),
        'social_profile_url': faker.url(),
        'bank_cards': (
            [
                faker.credit_card_number('visa16')
                for _ in range(faker.pyint(min_value=1, max_value=3))
            ]
            if faker.pybool
            else None
        ),
        'countries': ','.join(
            [faker.currency_code() for _ in range(faker.pyint(min_value=1, max_value=5))]
        ),
        'document': {
            'full_number': faker.pyint() if faker.pybool() else faker.pystr(),
        },
        'page_number': faker.pyint(min_value=0, max_value=10),
        'page_size': faker.pyint(min_value=1, max_value=100),
    }

И далее многократное выполнение. Пример кода для Pydantic:

def test_pydantic(data: dict) -> None:
    count = 10**5
    execution_time = timeit.timeit(stmt=lambda: PydanticSchema(**data), number=count)
    print('pydantic', count, execution_time)

Получаем такие результаты:

  1. Pydantic (6.85 сек)

  2. Trafaret (7.23 сек)

  3. Marshmallow (26.43 сек)

  4. DRF (36.42 сек)

Неожиданно, но Trafaret обгоняет Marshmallow на два корпуса и дышит лидеру в затылок. Первое и последнее место не принесли сюрпризов.

Если говорить о субъективном удобстве использования этих библиотек, то, на мой взгляд, порядок будет таким же.

Заключение

Изначально код для статьи писался в декабре 22 года. Что-то пошло не так, поэтому к этой идее я вернулся через год – в декабре 23. Сейчас уже весна 24, и я надеюсь, что прокрастинация всё же будет побеждена (а статья дописана), но благодаря ей можно посмотреть, как развивались обозреваемые библиотеки за год с небольшим.

Итоговые используемые версии библиотек
pydantic[email]==2.6.4
pydantic-extra-types==2.6.0
pycountry==23.12.11
marshmallow==3.21.1
marshmallow-union==0.1.15.post1
trafaret==2.1.1
djangorestframework==3.14.0

Например, за это время в Marshmallow завезли EnumUnion все еще приходится ставить дополнительно), а Pydantic обновил мажорную версию (вместе с ней и рекорды скорости). А вот у Trafaret за этот период вышло ровно одно обновление; вероятно, они достигли дзена.

Многие полезные фишки не рассматривались в рамках этой статьи, но о них можно почитать в документации. Такие как:

  • Использование нестандартной json библиотеки, а сторонних (например, orjson), которые обещают более высокую скорость сериализации и десериализации

  • Неизменяемость провалидированных данных (frozen)

  • Наследование/merge схем

Кроме рассмотренных библиотек существует еще множество других: Cerberus, jsonschema, WTForms, – но, полагаю, они уже в статусе легаси, поэтому в новые проекты не попадут.

P.S. если у вас есть что сказать об ошибках или улучшениях, смело пишите :)

Комментарии (3)


  1. Andrey_Solomatin
    18.03.2024 16:01

    Pydantic обновил мажорную версию (вместе с ней и рекорды скорости).

    Они переписали его на rust.

    А вот trafaret на питоне. https://github.com/Deepwalker/trafaret/


    1. klomytiz Автор
      18.03.2024 16:01

      Всё так. И это удивительно: trafaret, написанный на python, почти догоняет pydantic, который на rust. При этом первый ещё и выпускает по одному обновлению в год, а второй релизится раза по два каждый месяц

      Учитывая, что вряд ли временные затраты на валидацию являются хоть сколько-нибудь весомыми на фоне бизнес-логики и всяких IO-операций, то можно делать выбор библиотеки не по скорости, а по удобству использования


  1. andreymal
    18.03.2024 16:01

    По сути это не столько валидация, сколько парсинг — просто проверка того, что клиент прислал не совсем откровенный мусор. В реальности к этому добавляются примерно такие штуки, не рассмотренные в посте:

    • «email уже используется» или «выбранный объект не существует» — это вскользь упомянуто в посте, но не рассмотрено, а ведь сходить в базу (скорее всего асинхронно, 2024 год всё-таки) и проверить там наличие или отсутствие введённого значения всё равно так или иначе придётся;

    • валидация только при изменении — если значение не изменилось (опять же ходим в базу за старым значением), то свежедобавленные правила валидации не применяем, чтобы не злить старых пользователей;

      • более хитрый вариант — выполнить валидацию длины нового текста только если он длиннее старого (да, это реальная проверка, которую мне пришлось сделать на одном из моих сайтов);

    • локализация — не забываем, что тексты ошибок должны быть на родном пользователю языке и быть понятными ему (то есть, например, для встроенных валидаторов нужно заменить сухое «значение не соответствует регулярному выражению [A-Za-z0-9 ]+» на дружелюбное «допустимы только английские буквы, цифры и пробел»).

    Из знакомого мне обработать всё это более-менее внятно может разве что Django Forms, но это древнее зло, про которое никто уже не вспоминает, а в современных проектах как будто никто даже не пытается разруливать такие вещи — все тупо выполняют валидацию на фронте, а ошибки с бэка иногда вообще не выводят, и приходится смотреть мониторинг сети в девтулзах, чтобы понять, что я ввёл не так :(