Source code for caspia.meadow.value.validators

import logging
from datetime import datetime
from typing import Any, Callable, Optional

import arrow

Validator = Callable[[Any], bool]
logger = logging.getLogger(__name__)
_value_type_validators = dict()


def value_type_validator(value_type):
    def decorator(validator):
        _value_type_validators[value_type] = validator
        return validator

    return decorator


def get_validator(value_type) -> Optional[Validator]:
    return _value_type_validators.get(value_type)


def validate(value, value_type, optional=False, extra_validators=()):
    if optional and value is None:
        pass
    else:
        validator = get_validator(value_type)
        if validator is None:
            logger.warning('No validator for value type %r', value_type)
        else:
            validator(value)
        for validator in extra_validators:
            validator(value)


class MemberOf:
    def __init__(self, options):
        self.options = options

    def __call__(self, value):
        if value not in self.options:
            raise ValueError('Value %r is not member of %s' % (value, self.options))

    def description(self):
        return 'Allowed values: {}'.format(self.options)


class InRange:
    def __init__(self, start, end):
        self.start, self.end = start, end

    def __call__(self, value):
        if not self.start <= value <= (self.end + 0.0000000001):
            raise ValueError('Value %r is not in range [%s,%s]' % (value, self.start, self.end))

    def description(self):
        return 'Must belong to interval ({}, {}) (inclusive)'.format(self.start, self.end)


@value_type_validator('json')
def validate_json(_):
    pass


@value_type_validator('array')
def validate_array(value):
    if not isinstance(value, list):
        raise ValueError('%r is not a valid array' % value)


@value_type_validator('bool')
def validate_bool(value):
    if not isinstance(value, bool):
        raise ValueError('%r is not a valid bool' % value)


@value_type_validator('float')
def validate_float(value):
    if not isinstance(value, (float, int)):
        raise ValueError('%r is not a valid float' % value)


@value_type_validator('integer')
def validate_integer(value):
    if not isinstance(value, int):
        raise ValueError('%r is not a valid integer' % value)


@value_type_validator('string')
def validate_string(value):
    if not isinstance(value, str):
        raise ValueError('%r is not a valid string' % value)


@value_type_validator('void')
def validate_void(value):
    if value is not None:
        raise ValueError('%r is not a valid void' % value)


@value_type_validator('bytes')
def validate_bytes(value):
    if not isinstance(value, bytes):
        raise ValueError('%r is not a valid bytes value' % value)


@value_type_validator('date-time')
def validate_datetime(value):
    if not isinstance(value, (datetime, arrow.Arrow)):
        raise ValueError('%r is not a valid datetime obj' % value)