Intro

Есть такой шаг в развитии языка, когда его компилятор написан на нем же.
Чтобы доказать крутость библиотеки trafaret я тоже решил сделать что-то такое же
рекурсивненькое, где надо идти глубже.

Напишем на трафарете парсер Json Schema, который на выходе вернет
готовый трафарет для проверки документов в соответствии с данным описанием.

То есть некий объект типа Trafaret, если ему скормить корректный документ json schema
на выходе вернет объект типа Trafaret, которому можно кормить документы
соответствующие описанию.

Как вообще такое сделать библиотекой валидации? Ну библиотекой валидации никак, а вот универсальным трансформером запросто. Trafaret это парсер данных, точнее комбинаторный парсер. Только те парсеры про которые вы наслышаны такие как parsec и funcparserlib парсят строки, а Trafaret парсит все что под руку подвернется, и на что у автора талантов хватит закодировать.

Json Schema описывается как водится пачкой документов из которых самый близкий народу пожалуй этот — http://json-schema.org/latest/json-schema-validation.html
Тут описание множества ключевых слов с помощью которых вы можете описать критерии корректности документа, а вот замечательный и зубодробительный в реализации $ref всего в одном месте вскользь.

Часть первая

У json схемы в базовом варианте достаточно простая реализация — все ключевые слова, такие как maximum (максимальное значение для числа), pattern (регулярка по которой следует проверить строку), items (дочерная схема или массив схем для проверки элементов массива).
Так вот, все эти ключевые слова следует применять по отдельности. Встретили maximum, тут же проверим это число на соответствие верхней границе. То есть можно взять схему, например такую:

{
    "type": "number",
    "maximum": 45,
}

и разобрать её на составляющие, просто список проверок все из которых должны пройти.

validations = []
for key, value in schema:
    if key == 'type':
        if value == 'number':
            validations.append(is_a_number)

Ад какой, а еще схему бы проверить бы. Пожалуй закончим с наколенными примерами, начнем писать парсер. Json Schema это словарь, объект, мапа, короче ключи и значения в кудрявых скобках {}. Значит проверять будем словарь, попробуем:

import trafaret as t  # стандартный паттерн чтобы словом trafaret весь код не завалить

json_schema_type = t.Enum('null', 'boolean', 'object', 'array', 'number', 'integer', 'string')

json_schema_keys = t.Dict(
    t.Key('type', optional=True, trafaret=json_schema_type),
    t.Key('maximum', optional=True, trafaret=t.Int()),
)

Мы взяли только пару ключевых слов чтобы не утомлять читателя прокруткой. Валидация сработает, но нам надо в итоге не только проверить схему, но и получить валидаторы. А это как раз то, что trafaret в отличие от многих делает на раз два, но придется немного подумать.

Есть операция &, берет два трафарета и применяет второй к выходу первого если не было
ошибок валидации, то есть типа так:

check_password = (
    t.String() & (lambda value: value if value == 'secret' else t.DataError('Wrong password'))
)

Если на вход передать не строку check_password(123), то на выходе мы сразу получим сообщение о том, что значение не строка и проверять на соотвествие строке 'secret' не будем.
Чтобы проверять любые значения python на равенство в трафарете есть Atom.

И можно было бы типы описать вроде как:

json_schema_type = (
    t.Atom('null') & t.Null()
    | t.Atom('boolean') & t.Bool()
)

Но это немного не то, чего мы хотим. Мы хотим вернуть трафарет и не применить его тут же
с заведомо ошибочным вариантом — строка 'null' точно не None.

Напишем хелпер, который тоже трафарет, и возвращает заданный трафарет:

def just(trafaret):
    """Returns trafaret and ignoring values"""
    def create(value):
        return trafaret
    return create

И применим:

json_schema_type = (
    t.Atom('null') & just(t.Null())
    | t.Atom('boolean') & just(t.Bool())
    | t.Atom('object') & just(t.Type(dict))
    | t.Atom('array') & just(t.Type(list))
    | t.Atom('number') & just(t.Float())
    | t.Atom('integer') & just(t.Int())
    | t.Atom('string') & just(t.String())
)

Теперь вызов json_schema_type('null') вернет нам экземпляр t.Null(). Парсер стал порождать
итоговый результат.

Первый уровень сложности пройден, мы реализовали type. С радостью тем же образом делаем enum, const, multipleOf, maximum etc.

Переходим ко второму уровню сложности

Практически все ключевики в json схеме независимы, но часть все таки зависит друг от друга. Это ключевые слова для массивов и объектов. additionalItems это дочерняя схема для проверки элементов массива, которые не описаны в items. То есть например "items": [{"type":"string"}, {"type":"bool"}] проверит первые два элемента, а вот если в проверяемом документе их окажется 3 и более, то проверять их следует уже через additionalItems, если они заданы, или же
это уже ошибка само по себе.

Второй случай — additionalProperties. Для проверки объектов в json схема используются properties и patternProperties, а для всего что не описано в первых двух применяется additionalProperties.

Это в принципе давно решеная тема в трафаретостроении, используются особые ключи, но трафаретостроением заняты еще не 100% населения, так что остановимся чуть подробнее.

Чтобы проверять словари в трафарете использован не совсем стандартный подход. По сути
проверкой словарей в трафарете занимаются как раз ключи, в частности Key, а сам Dict
это обвязка вокруг которая собирает результаты исполнения всех ключей на данном объекте.

Тип ключа в терминах mypy выглядит так:

KeyT = Callable[  # некая функция или класс с __call__
    [Mapping],  # принимает один аргумент с интерфейсом Mapping (т.е. dict подходит)
    Sequence[  # возвращает последовательность, то есть ключи на деле генераторы
        Tuple[  # а каждый элемент последовательности это кортеж из трех элементов
            str,  # имя ключа которое надо включить в итоговый объект – может и не быть исходным если будет переименование
            Union[Any, t.DataError],  # тут или какое-то значение или ошибка трафарета DataError
            Sequence[str]  # ключ козвращает имена всех ключей которые он дергал в процессе своей работы
        ]
    ]
]

Смотрим внимательно три раза и обращаем внимание на последнюю строчку перед скобкопадом — ключ рапортует обо всех ключах словаря которые он дергал. То есть Ключ может дернуть кучу ключей, а также вернуть любое число ключей. Знать какие ключи словаря дергал Ключ надо чтобы узнать есть ли в словаре лишние или дополнительные элементы.

Отсюда следует что ключ может взять сразу пачку ключей и проверить их разом. Это вообще было решение что делать с password & password_confirmation когда ключи сами себе такие независимые. Но в нашем случае задача несколько хитрее чем сравнить два ключа на равенство, а заложенная гибкость еще и не то позволяет сделать.

Встречаем subdict:

def subdict(name, *keys, **kw):
    trafaret = kw.pop('trafaret')  # coz py2k

    def inner(data, context=None):
        errors = False
        preserve_output = []  # если будут ошибки, надо их все отсюда зарапортовать
        touched = set()
        collect = {}
        for key in keys:
            for k, v, names in key(data, context=context):
                touched.update(names)
                preserve_output.append((k, v, names))
                if isinstance(v, t.DataError):
                    errors = True
                else:
                    collect[k] = v
        if errors:
            for out in preserve_output:
                yield out
        elif collect:  # в случае успеха всю пачку ключей со значениями можем передать более общему трафарету
            yield name, t.catch(trafaret, **collect), touched

    return inner

И примерно так он применен в недрах trafaret_schema:

subdict(
    'array',
    t.Key('items', optional=True, trafaret=ensure_list(json_schema)),
    t.Key('additionalItems', optional=True, trafaret=json_schema),
    trafaret=check_array,
),

Фигура третья, нужен стейт

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

Но на следующем уровне json схемы нас встречает мега босс — $ref. Очень разумная штука, позволяет ссылаться на уже определенную где-то другую схему. Например схема может быть определена в definitions или вообще в другом документе.

А значит в процессе разбора схемы все определения схема с их адресами нам надо собрать в одном месте. Потом проверить, что все $ref встреченные в документе имеют определения. А в процессе выполнения уже понятно — трафарет для $ref просто дернет нужный трафарет из реестра.

Ну хорошо, написать реестр раз плюнуть:

class Register:
    def __init__(self):
        "Аз есмь реестр"
        pass

А вот дальше пришлось трафарет доработать и помимо стандартного аргумента value можно теперь пустить по цепочке в стандартные трафареты еще и context=Any. А собственно наш только что написанные Register и есть этот самый контекст.

Применяем примерно так для определения трафарета для $ref:

def ref_field(reference, context=None):
    register = context  # просто для читаемости называем context нормально
    register.reg_reference(reference)  # регистрируем своей значение $ref, чтобы можно было проверить целостность ссылок

    def inner(value, context=None):  # то все были замыкания, а вот и трафарет
        schema = register.get_schema(reference)  # получаем из реестра по ссылке наш трафарет
        return schema(value, context=context)  # проверяем с его помощью значение

    return inner

Конечно самая морока это собрать ссылки на все возможное подсхемы.
Вот пример для дочерней схемы ссылку на которую хочется сохранить:

def deep_schema(key):  # запоминаем ключик
    def inner(data, context=None):
        register = context
        register.push(key)  # этот самый key это часть пути до нашей дочерней схемы
        # и собственно вся магия тут — все дочерние схемы тоже сделают push, и в реестре сохранится
        # путь до каждого ключа
        try:
            schema = json_schema(data, context=register)
            register.save_schema(schema)  # если схема валидна, реестр запоминает нашу схему по данному пути
            return schema
        finally:
            register.pop()
    return t.Call(inner)

Зачем

Лучшее от двух миров — json схема широко распространена и поддерживается любыми языками. А трафарет лучшая библиотека трансформаций с проверками под python. Точнее единственная. И главное, для ключевого слова format можно подсунуть любой трафарет примерно так:

import trafaret as t
from trafaret_schema import json_schema, Register

my_reg = Register()

my_reg.reg_format('any_ip', t.IPv4 | t.IPv6)

check_address = json_schema(open('address.rjson').read(), context=register)
check_person = json_schema(open('person.json').read(), context=register)

Итоги

trafaret_schema работает, готова к использованию, пишите если что не так, будем править. Смотрим https://github.com/Deepwalker/trafaret_schema или pip install trafaret_schema.
В процессе Trafaret получил поддержку контекстов, а заодно уж и async/await, чтобы
в случае asyncio было что с этим контекстом делать.
Мы получили этот замечательный текст как на трафаретах делать все что угодно, в том числе и другие трафареты.

Trafaret как парсер. Реализация JSON Schema