Source code for quantopian_tools.schema

# -*- coding: utf-8 -*-
from __future__ import print_function, absolute_import, division, unicode_literals

import datetime

import cerberus
import six

from quantopian_tools.exceptions import SchemaValidationError


def validate(data, schema, raise_exc=False, **kwargs):
    validator = CustomValidator(schema, **kwargs)
    if not validator.validate(data):
        if raise_exc:
            raise SchemaValidationError(data, schema, validator.errors)
        return False, validator.errors
    if raise_exc:
        return validator.document
    return True, validator.document


[docs]class CustomValidator(cerberus.Validator):
    def _normalize_rename_handler(self, mapping, schema, field):
        """ {'oneof': [
                {'type': 'callable'},
                {'type': 'list',
                 'schema': {'oneof': [{'type': 'callable'},
                                      {'type': 'string'}]}},
                {'type': 'string'}
                ]} """
        if 'rename' in schema[field]:
            new_field = schema[field]['rename']
            schema[new_field] = schema[field]
            del schema[field]
            field = new_field
        super(CustomValidator, self)._normalize_rename_handler(mapping, schema, field)

    # Custom coerce functions
    def _normalize_coerce_millis_timestamp(self, value):
        if value is None:
            return None
        if isinstance(value, datetime.datetime):
            return value
        elif isinstance(value, int):
            return datetime.datetime.fromtimestamp(value / 1000.0)
        else:
            raise ValueError("cannot coerce type {} to a datetime".format(type(value).__name__))

    def _normalize_coerce_datetime_to_date(self, value):
        if value is None:
            return None
        if isinstance(value, datetime.datetime):
            return value.date()
        else:
            raise ValueError("cannot coerce type {} to a date".format(type(value).__name__))

    def _normalize_coerce_int(self, value):
        if value is None:
            return None
        elif isinstance(value, six.integer_types + six.string_types + (float,)):
            try:
                return int(value)
            except Exception as ex:
                raise ValueError(str(ex))
        raise ValueError("cannot coerce type {} to an int".format(type(value).__name__))

    def _normalize_coerce_number(self, value):
        if value is None:
            return None
        elif isinstance(value, six.integer_types + (float,)):
            return value
        elif isinstance(value, six.string_types):
            try:
                try:
                    return int(value)
                except ValueError:
                    return float(value)
            except Exception as ex:
                raise ValueError(str(ex))
        raise ValueError("cannot coerce type {} to a number".format(type(value).__name__))

    def _normalize_coerce_bool(self, value):
        if value is None:
            return False
        elif isinstance(value, bool):
            return value
        elif isinstance(value, six.string_types) and value.lower().strip() in ['true', '1']:
            return True
        elif isinstance(value, six.string_types) and value.lower().strip() in ['', 'false', '0']:
            return False
        raise ValueError("cannot coerce type {} to a bool".format(type(value).__name__))

    def _normalize_coerce_str(self, value):
        if value is None:
            return None
        if not isinstance(value, six.string_types):
            value = str(value)
        return value

    def _normalize_coerce_strip(self, value):
        if value is None:
            return None
        if hasattr(value, 'strip') and callable(getattr(value, 'strip')):
            return value.strip()
        raise ValueError("cannot coerce type {}; no strip() method found".format(type(value).__name__))

    def _normalize_coerce_filter_falsey(self, value):
        if value is None:
            return None
        if isinstance(value, (list, tuple, set)):
            return [item for item in value if item]
        raise ValueError("cannot coerce type {} to a list".format(type(value).__name__))

    def _normalize_coerce_falsey_to_none(self, value):
        return value or None

    def _normalize_coerce_first_item(self, value):
        if value is None:
            return None
        if isinstance(value, (list, tuple, set)):
            if len(value) >= 1:
                return value[0]
            return None
        raise ValueError("cannot get first item of type {}".format(type(value).__name__))


# Schema helpers

def string(required=False, nullable=False, empty=True, strip=False, **kwargs):
    schema = {
        'type': 'string',
        'required': required,
        'nullable': nullable,
        'empty': empty,
        'coerce': 'str' if not strip else ('str', 'strip')
    }
    schema.update(kwargs)
    return schema


def boolean(required=False, nullable=False, **kwargs):
    schema = {
        'type': 'boolean',
        'required': required,
        'nullable': nullable,
        'coerce': 'bool'
    }
    schema.update(kwargs)
    return schema


def datetime_(required=False, nullable=False, **kwargs):
    schema = {
        'type': 'datetime',
        'required': required,
        'nullable': nullable
    }
    schema.update(kwargs)
    return schema


def date_(required=False, nullable=False, **kwargs):
    schema = {
        'type': 'date',
        'required': required,
        'nullable': nullable
    }
    schema.update(kwargs)
    return schema


def integer(required=False, nullable=False, **kwargs):
    schema = {
        'type': 'integer',
        'required': required,
        'nullable': nullable,
        'coerce': 'int'
    }
    schema.update(kwargs)
    return schema


def number(required=False, nullable=False, **kwargs):
    schema = {
        'type': 'number',
        'required': required,
        'nullable': nullable,
        'coerce': 'number'
    }
    schema.update(kwargs)
    return schema


def dictionary(required=False, nullable=False, **kwargs):
    schema = {
        'type': 'dict',
        'required': required,
        'nullable': nullable
    }
    schema.update(kwargs)
    return schema


def list_(required=False, nullable=False, **kwargs):
    schema = {
        'type': 'list',
        'required': required,
        'nullable': nullable
    }
    schema.update(kwargs)
    return schema