schema_with_class.py

# -*- coding:utf-8 -*-
class ValidationError(Exception):
    pass


class Field(object):
    def __init__(self, required=True):
        self.required = required

    def __call__(self, value):
        try:
            if self.required is False and value is None:
                return None
            self.pre_validate(value)
            parsed = self.parse(value)
            self.post_validate(parsed)
            return parsed
        except Exception as e:
            raise ValidationError(e)

    def pre_validate(self, value):
        return value

    def parse(self, value):
        return value

    def post_validate(self, value):
        return value


class IntegerField(Field):
    def parse(self, value):
        return int(value)


class PositiveIntegerField(IntegerField):
    def post_validate(self, value):
        super(IntegerField, self).post_validate(value)
        assert value >= 0


assert PositiveIntegerField()("10") == 10

# validation error
# assert PositiveIntegerField()("-10") == 10

assert PositiveIntegerField(required=False)(None) is None

schema_with_function.py

# -*- coding:utf-8 -*-
class ValidationError(Exception):
    pass


class Break(Exception):
    def __init__(self, value):
        self.value = value


def parse(value, convertions, options):
    try:
        for cnv in convertions:
            value = cnv(value, options)
        return value
    except Break as e:
        return e.value
    except Exception as e:
        raise ValidationError(e)


class merged(object):
    def __init__(self, fn, *args, **kwargs):
        self.fn = fn
        self.args = list(args)
        self.kwargs = kwargs

    def merged(self, pre=None, post=None, **kwargs_):
        args = self.args[:]
        if pre:
            args.insert(0, pre)
        elif post:
            args.append(post)
        kwargs = self.kwargs.copy()
        kwargs.update(kwargs_)
        return merged(self.fn, *args, **kwargs)

    def __call__(self, value):
        return self.fn(value, self.args, self.kwargs)


def reject_None(val, options):
    if options.get("required", True):
        return val
    raise Break(None)


def parse_int(val, options):
    return int(val)


def validate_positive(val, option):
    assert val >= 0
    return val


Field = merged(parse).merged
IntegerField = Field(pre=reject_None, post=parse_int).merged
PositiveIntegerField = IntegerField(post=validate_positive).merged


assert PositiveIntegerField()(10) == 10

# assert PositiveIntegerField()(-10) == 10 # raise ValidationError

assert PositiveIntegerField(required=False)(None) is None