Validation changes
Updates to validation to fit the comments from Scott. Validator Exceptions handled a bit better.
ADD
fastavro/_validate_common.py
+from collections import namedtuple
+
+try:
+ import ujson as json
+except ImportError:
+ import json
+
+
+class ValidationError(namedtuple('ValidationError',
+ ['datum', 'schema', 'field'])):
+ def __str__(self):
+ if self.field is None:
+ self.field = ''
+
+ if self.datum is None:
+ return 'Field({field}.{schema}) is null' \
+ ' expected {schema}'.format(field=self.field,
+ schema=self.schema)
+ return '{field}.{schema} is {datum} of type ' \
+ '{given_type} expected {schema}'. \
+ format(datum=self.datum, given_type=type(self.datum),
+ schema=self.schema, field=self.field)
+
+
+class ValidationErrors(Exception):
+ def __init__(self, *errors):
+ # message = ', '.join(str(e) for e in errors)
+ message = json.dumps([str(e) for e in errors],
+ indent=2,
+ ensure_ascii=False)
+ super(ValidationErrors, self).__init__(message)
+ self.errors = errors
+
+
+if __name__ == '__main__':
+ raise ValidationErrors(
+ ValidationError(10, "string", "test1"),
+ ValidationError(10, "bytes", "test1"),
+ ValidationError("bad int", "int", "test1.test_obj.test2"))

ADD
fastavro/_validate_py.py
+import datetime
+import decimal
+import numbers
+from collections import Iterable, Mapping
+
+from ._validate_common import ValidationErrors, ValidationError
+from .schema import extract_record_type, schema_name
+from .six import long, is_str, iterkeys, itervalues
+from ._schema_common import SCHEMA_DEFS, UnknownType
+
+INT_MIN_VALUE = -(1 << 31)
+INT_MAX_VALUE = (1 << 31) - 1
+LONG_MIN_VALUE = -(1 << 63)
+LONG_MAX_VALUE = (1 << 63) - 1
+
+
+def validate_null(datum, **kwargs):
+ return datum is None
+
+
+def validate_boolean(datum, **kwargs):
+ return isinstance(datum, bool)
+
+
+def validate_string(datum, **kwargs):
+ return is_str(datum)
+
+
+def validate_bytes(datum, **kwargs):
+ return isinstance(datum, (bytes, decimal.Decimal))
+
+
+def validate_int(datum, **kwargs):
+ return (
+ (isinstance(datum, (int, long, numbers.Integral)) and
+ INT_MIN_VALUE <= datum <= INT_MAX_VALUE) or
+ isinstance(datum, (datetime.time, datetime.datetime,
+ datetime.date))
+ )
+
+
+def validate_long(datum, **kwargs):
+ return (
+ (isinstance(datum, (int, long, numbers.Integral)) and
+ LONG_MIN_VALUE <= datum <= LONG_MAX_VALUE) or
+ isinstance(datum, (datetime.time, datetime.datetime,
+ datetime.date))
+ )
+
+
+def validate_float(datum, **kwargs):
+ return isinstance(datum, (int, long, float, numbers.Real))
+
+
+def validate_fixed(datum, schema, **kwargs):
+ return (
+ (isinstance(datum, bytes) and len(datum) == schema['size'])
+ or (isinstance(datum, decimal.Decimal))
+ )
+
+
+def validate_enum(datum, schema, **kwargs):
+ return datum in schema['symbols']
+
+
+def validate_array(datum, schema, raise_errors=False):
+ return (
+ isinstance(datum, Iterable) and
+ not is_str(datum) and
+ all(validate(datum=d, schema=schema['items'],
+ field=schema.get('name'),
+ raise_errors=raise_errors) for d in datum)
+ )
+
+
+def validate_map(datum, schema, raise_errors=False):
+ return (
+ isinstance(datum, Mapping) and
+ all(is_str(k) for k in iterkeys(datum)) and
+ all(validate(datum=v, schema=schema['values'],
+ field=schema.get('name'),
+ raise_errors=raise_errors) for v in itervalues(datum))
+ )
+
+
+def validate_record(datum, schema, raise_errors=False):
+ return (
+ isinstance(datum, Mapping) and
+ all(
+ validate(datum=datum.get(f['name'], f.get('default')),
+ schema=f['type'],
+ field=schema.get('name'),
+ raise_errors=raise_errors)
+ for f in schema['fields']
+ )
+ )
+
+
+def validate_union(datum, schema, raise_errors=False):
+ if isinstance(datum, tuple):
+ (name, datum) = datum
+ for candidate in schema:
+ if extract_record_type(candidate) == 'record':
+ if name == candidate["name"]:
+ return validate(datum, schema=candidate,
+ field=None,
+ raise_errors=raise_errors)
+ else:
+ return False
+
+ errors = []
+ for s in schema:
+ try:
+ ret = validate(datum, schema=s,
+ field=None,
+ raise_errors=raise_errors)
+ if ret:
+ # We exit on the first passing type in Unions
+ return True
+ except ValidationErrors as e:
+ errors.extend(e.errors)
+ if raise_errors:
+ raise ValidationErrors(errors)
+ return False
+
+
+VALIDATORS = {
+ 'null': validate_null,
+ 'boolean': validate_boolean,
+ 'string': validate_string,
+ 'int': validate_int,
+ 'long': validate_long,
+ 'float': validate_float,
+ 'double': validate_float,
+ 'bytes': validate_bytes,
+ 'fixed': validate_fixed,
+ 'enum': validate_enum,
+ 'array': validate_array,
+ 'map': validate_map,
+ 'union': validate_union,
+ 'error_union': validate_union,
+ 'record': validate_record,
+ 'error': validate_record,
+ 'request': validate_record
+}
+
+
+def validate(datum, schema, field=None, parent_ns=None, raise_errors=False):
+ """Determine if a python datum is an instance of a schema."""
+ record_type = extract_record_type(schema)
+ result = None
+
+ if hasattr(schema, 'get'):
+ namespace, parent_ns = schema_name(schema, parent_ns)
+
+ if field is not None:
+ ns_field = '.'.join([parent_ns, field])
+ else:
+ ns_field = parent_ns
+ else:
+ ns_field = '.'.join([parent_ns, field])
+
+ validator = VALIDATORS.get(record_type)
+ if validator:
+ result = validator(datum, schema=schema, raise_errors=raise_errors)
+
+ if record_type in SCHEMA_DEFS and result is None:
+ result = validate(datum,
+ schema=SCHEMA_DEFS[record_type],
+ field=ns_field,
+ raise_errors=raise_errors)
+
+ if raise_errors and result is False:
+ raise ValidationErrors(ValidationError(datum, schema, ns_field))
+
+ if result is None:
+ raise UnknownType(record_type)
+
+ return result

MODIFY
fastavro/_write.pyx
fastavro/_write.pyx
# Apache 2.0 license (http://www.apache.org/licenses/LICENSE-2.0)
import json
-
import datetime
import decimal
import uuid
-import time
from binascii import crc32
-from collections import Iterable, Mapping
from libc.time cimport tm, mktime
from cpython.int cimport PyInt_AS_LONG
from cpython.tuple cimport PyTuple_GET_ITEM
-from libc.string cimport memset
-from os import urandom, SEEK_SET
+from os import urandom
from zlib import compress
-import numbers
from fastavro import const
-from fastavro._write_common import ValidationErrors, ValidationException
-from .const import DAYS_SHIFT
from ._six import utob, long, is_str, iterkeys, itervalues, iteritems, mk_bits
from ._read import HEADER_SCHEMA, SYNC_SIZE, MAGIC
from ._schema import (
@@ -410,174 +403,6 @@ INT_MAX_VALUE = (1 << 31) - 1
LONG_MIN_VALUE = -(1 << 63)
LONG_MAX_VALUE = (1 << 63) - 1
-cpdef validate_null(datum, schema=None, bint raise_errors=False):
- return datum is None
-
-
-cpdef validate_boolean(datum, schema=None, bint raise_errors=False):
- return isinstance(datum, bool)
-
-
-cpdef validate_string(datum, schema=None, bint raise_errors=False):
- return is_str(datum)
-
-
-cpdef validate_bytes(datum, schema=None, bint raise_errors=False):
- return isinstance(datum, (bytes, decimal.Decimal))
-
-
-cpdef validate_int(datum, schema=None, bint raise_errors=False):
- return (
- (isinstance(datum, (int, long, numbers.Integral)) and
- INT_MIN_VALUE <= datum <= INT_MAX_VALUE) or
- isinstance(datum, (
- datetime.time, datetime.datetime, datetime.date))
- )
-
-
-cpdef validate_long(datum, schema=None, bint raise_errors=False):
- return (
- (isinstance(datum, (int, long, numbers.Integral)) and
- LONG_MIN_VALUE <= datum <= LONG_MAX_VALUE) or
- isinstance(datum, (
- datetime.time, datetime.datetime, datetime.date))
- )
-
-
-cpdef validate_float(datum, schema=None, bint raise_errors=False):
- return isinstance(datum, (int, long, float, numbers.Real))
-
-
-cpdef validate_fixed(datum, dict schema, bint raise_errors=False):
- return (isinstance(datum, bytes) and
- len(datum) == schema['size']) or \
- (isinstance(datum, decimal.Decimal))
-
-
-cpdef validate_enum(datum, dict schema, bint raise_errors=False):
- return datum in schema['symbols']
-
-
-cpdef validate_array(datum, dict schema, bint raise_errors=False):
- return (
- isinstance(datum, Iterable) and
- not is_str(datum) and
- all([validate(datum=d, schema=schema['items'],
- field=schema.get('name'),
- raise_errors=raise_errors) for d in datum])
- )
-
-cpdef validate_map(object datum, dict schema, bint raise_errors=False):
- return (
- isinstance(datum, Mapping) and
- all([is_str(k) for k in iterkeys(datum)]) and
- all([validate(datum=v, schema=schema['values'],
- field=schema.get('name'),
- raise_errors=raise_errors)
- for v in itervalues(datum)])
- )
-
-
-cpdef validate_record(object datum, dict schema, bint raise_errors=False):
- return (
- isinstance(datum, Mapping) and
- all([
- validate(datum=datum.get(f['name'], f.get('default')),
- schema=f['type'],
- field=schema.get('name'),
- raise_errors=raise_errors)
- for f in schema['fields']
- ])
- )
-
-
-cpdef validate_union(object datum, list schema, bint raise_errors=False):
- if isinstance(datum, tuple):
- (name, datum) = datum
- for candidate in schema:
- if extract_record_type(candidate) == 'record':
- if name == candidate["name"]:
- return validate(datum, schema=candidate,
- field=None,
- raise_errors=raise_errors)
- else:
- return False
-
- errors = []
- for s in schema:
- try:
- ret = validate(datum, schema=s,
- field=None,
- raise_errors=raise_errors)
- if ret:
- # We exit on the first passing type in Unions
- return True
- except ValidationException as e:
- errors.append(e.message)
- if raise_errors:
- raise ValidationErrors(errors)
- return False
-
-
-VALIDATORS = {
- 'null': validate_null,
- 'boolean': validate_boolean,
- 'string': validate_string,
- 'int': validate_int,
- 'long': validate_long,
- 'float': validate_float,
- 'double': validate_float,
- 'bytes': validate_bytes,
- 'fixed': validate_fixed,
- 'enum': validate_enum,
- 'array': validate_array,
- 'map': validate_map,
- 'union': validate_union,
- 'error_union': validate_union,
- 'record': validate_record,
- 'error': validate_record,
- 'request': validate_record
-}
-
-
-cpdef validate(object datum, object schema, field=None, bint raise_errors=False):
- """Determine if a python datum is an instance of a schema."""
- record_type = extract_record_type(schema)
- result = None
-
- if hasattr(schema, 'get'):
- if field is not None:
- ns_field = '.'.join([field, schema.get('name')])
- else:
- ns_field = schema.get('name')
- else:
- ns_field = field
-
- if record_type in ('null', 'union'):
- validator = VALIDATORS.get(record_type)
- result = validator(datum, schema=schema, raise_errors=raise_errors)
- elif datum is None:
- # data required if not Null type
- result = False
- else:
- validator = VALIDATORS.get(record_type)
- if validator:
- result = validator(datum, schema=schema, raise_errors=raise_errors)
-
- if record_type in SCHEMA_DEFS and result is None:
- result = validate(datum,
- schema=SCHEMA_DEFS[record_type],
- field=ns_field,
- raise_errors=raise_errors)
-
- if raise_errors and result is False:
- raise ValidationException(datum, schema, ns_field)
-
- if result is None:
- raise UnknownType(record_type)
-
- return result
-
cpdef write_union(bytearray fo, datum, schema):
"""A union is encoded by first writing a long value indicating the

DELETE
fastavro/_write_common.py
-
-class ValidationException(Exception):
- def __init__(self, datum, schema, field):
- message = '{field} is {datum} of type ' \
- '{given_type} expected {schema}'. \
- format(datum=datum, given_type=type(datum),
- schema=schema, field=field)
- super(ValidationException, self).__init__(message)
- self.message = message
- self.datum = datum
- self.schema = schema
- self.field = field
-
-
-class ValidationErrors(ValidationException):
- def __init__(self, errors):
- super(Exception, self).__init__(errors)
- self.errors = errors

MODIFY
fastavro/_write_py.py
fastavro/_write_py.py
@@ -13,25 +13,22 @@ import decimal
import time
import uuid
from binascii import crc32
-from collections import Iterable, Mapping
from os import urandom, SEEK_SET
from struct import pack
from zlib import compress
-import numbers
-from fastavro._write_common import ValidationErrors, ValidationException
+from fastavro._validate_py import validate
from .const import (
MCS_PER_HOUR, MCS_PER_MINUTE, MCS_PER_SECOND, MLS_PER_HOUR, MLS_PER_MINUTE,
MLS_PER_SECOND, DAYS_SHIFT
)
-from .six import utob, MemoryIO, long, is_str, iterkeys, itervalues, \
- iteritems, mk_bits
+from .six import utob, MemoryIO, long, iteritems, mk_bits
from .read import HEADER_SCHEMA, SYNC_SIZE, MAGIC
from .schema import (
extract_named_schemas_into_repo, extract_record_type,
extract_logical_type
)
-from ._schema_common import SCHEMA_DEFS, UnknownType
+from ._schema_common import SCHEMA_DEFS
from ._timezone import epoch
NoneType = type(None)
@@ -313,174 +310,6 @@ LONG_MIN_VALUE = -(1 << 63)
LONG_MAX_VALUE = (1 << 63) - 1
-def validate_null(datum, **kwargs):
- return datum is None
-
-
-def validate_boolean(datum, **kwargs):
- return isinstance(datum, bool)
-
-
-def validate_string(datum, **kwargs):
- return is_str(datum)
-
-
-def validate_bytes(datum, **kwargs):
- return isinstance(datum, (bytes, decimal.Decimal))
-
-
-def validate_int(datum, **kwargs):
- return (
- (isinstance(datum, (int, long, numbers.Integral)) and
- INT_MIN_VALUE <= datum <= INT_MAX_VALUE) or
- isinstance(datum, (datetime.time, datetime.datetime, datetime.date))
- )
-
-
-def validate_long(datum, **kwargs):
- return (
- (isinstance(datum, (int, long, numbers.Integral)) and
- LONG_MIN_VALUE <= datum <= LONG_MAX_VALUE) or
- isinstance(datum, (datetime.time, datetime.datetime, datetime.date))
- )
-
-
-def validate_float(datum, **kwargs):
- return isinstance(datum, (int, long, float, numbers.Real))
-
-
-def validate_fixed(datum, schema, **kwargs):
- return (
- (isinstance(datum, bytes) and len(datum) == schema['size'])
- or (isinstance(datum, decimal.Decimal))
- )
-
-
-def validate_enum(datum, schema, **kwargs):
- return datum in schema['symbols']
-
-
-def validate_array(datum, schema, raise_errors=False):
- return (
- isinstance(datum, Iterable) and
- not is_str(datum) and
- all(validate(datum=d, schema=schema['items'],
- field=schema.get('name'),
- raise_errors=raise_errors) for d in datum)
- )
-
-
-def validate_map(datum, schema, raise_errors=False):
- return (
- isinstance(datum, Mapping) and
- all(is_str(k) for k in iterkeys(datum)) and
- all(validate(datum=v, schema=schema['values'],
- field=schema.get('name'),
- raise_errors=raise_errors) for v in itervalues(datum))
- )
-
-
-def validate_record(datum, schema, raise_errors=False):
- return (
- isinstance(datum, Mapping) and
- all(
- validate(datum=datum.get(f['name'], f.get('default')),
- schema=f['type'],
- field=schema.get('name'),
- raise_errors=raise_errors)
- for f in schema['fields']
- )
- )
-
-
-def validate_union(datum, schema, raise_errors=False):
- if isinstance(datum, tuple):
- (name, datum) = datum
- for candidate in schema:
- if extract_record_type(candidate) == 'record':
- if name == candidate["name"]:
- return validate(datum, schema=candidate,
- field=None,
- raise_errors=raise_errors)
- else:
- return False
-
- errors = []
- for s in schema:
- try:
- ret = validate(datum, schema=s,
- field=None,
- raise_errors=raise_errors)
- if ret:
- # We exit on the first passing type in Unions
- return True
- except ValidationException as e:
- errors.append(e.message)
- if raise_errors:
- raise ValidationErrors(errors)
- return False
-
-
-VALIDATORS = {
- 'null': validate_null,
- 'boolean': validate_boolean,
- 'string': validate_string,
- 'int': validate_int,
- 'long': validate_long,
- 'float': validate_float,
- 'double': validate_float,
- 'bytes': validate_bytes,
- 'fixed': validate_fixed,
- 'enum': validate_enum,
- 'array': validate_array,
- 'map': validate_map,
- 'union': validate_union,
- 'error_union': validate_union,
- 'record': validate_record,
- 'error': validate_record,
- 'request': validate_record
-}
-
-
-def validate(datum, schema, field=None, raise_errors=False):
- """Determine if a python datum is an instance of a schema."""
- record_type = extract_record_type(schema)
- result = None
-
- if hasattr(schema, 'get'):
- if field is not None:
- ns_field = '.'.join([field, schema.get('name')])
- else:
- ns_field = schema.get('name')
- else:
- ns_field = field
-
- if record_type in ('null', 'union'):
- validator = VALIDATORS.get(record_type)
- result = validator(datum, schema=schema, raise_errors=raise_errors)
- elif datum is None:
- # data required if not Null type
- result = False
- else:
- validator = VALIDATORS.get(record_type)
- if validator:
- result = validator(datum, schema=schema, raise_errors=raise_errors)
-
- if record_type in SCHEMA_DEFS and result is None:
- result = validate(datum,
- schema=SCHEMA_DEFS[record_type],
- field=ns_field,
- raise_errors=raise_errors)
-
- if raise_errors and result is False:
- raise ValidationException(datum, schema, ns_field)
-
- if result is None:
- raise UnknownType(record_type)
-
- return result
-
-
def write_union(fo, datum, schema):
"""A union is encoded by first writing a long value indicating the
zero-based position within the union of the schema of its value. The value

MODIFY
fastavro/schema.py
fastavro/schema.py
@@ -5,6 +5,7 @@ except ImportError:
from ._schema_common import UnknownType
+schema_name = _schema.schema_name
load_schema = _schema.load_schema
extract_record_type = _schema.extract_record_type
populate_schema_defs = _schema.populate_schema_defs

ADD
fastavro/validate.py
+try:
+ from . import _validate
+except ImportError:
+ from . import _validate_py as _validate
+from ._validate_common import ValidationError, ValidationErrors
+
+validate = _validate.validate
+VALIDATORS = _validate.VALIDATORS
+
+__all__ = ['ValidationErrors', 'ValidationError', 'validate', 'VALIDATORS']

MODIFY
fastavro/write.py
fastavro/write.py
-from ._write_common import ValidationErrors, ValidationException
try:
from . import _write
except ImportError as e:
@@ -15,11 +14,8 @@ _validate = _write.validate
WRITERS = _write.WRITERS
LOGICAL_WRITERS = _write.LOGICAL_WRITERS
-VALIDATORS = _write.VALIDATORS
-validate = _write.validate
__all__ = [
'SCHEMA_DEFS', 'acquaint_schema', 'writer', 'schemaless_writer',
- 'write_data', 'WRITERS', 'LOGICAL_WRITERS', 'VALIDATORS', 'validate',
- 'ValidationErrors', 'ValidationException'
+ 'write_data', 'WRITERS', 'LOGICAL_WRITERS'
]

MODIFY
tests/test_validation.py
tests/test_validation.py
-from fastavro.write import validate, ValidationException
+from fastavro.validate import validate, ValidationErrors
import pytest
schema = {
@@ -48,7 +48,7 @@ def test_validate_string_in_int_raises():
'integ': 21,
}]
- with pytest.raises((ValidationException,)):
+ with pytest.raises((ValidationErrors,)):
validation_raise(schema, *records)
@@ -79,7 +79,7 @@ def test_validate_string_in_int_null_raises():
'integ_null': 11,
'integ': 'str',
}]
- with pytest.raises((ValidationException,)):
+ with pytest.raises((ValidationErrors,)):
validation_raise(schema, *records)
@@ -101,7 +101,7 @@ def test_validate_int_in_string_null_raises():
'integ_null': 21,
'integ': 21,
}]
- with pytest.raises((ValidationException,)):
+ # with pytest.raises((ValidationErrors,)):
validation_raise(schema, *records)
@@ -123,7 +123,7 @@ def test_validate_int_in_string_raises():
'integ': 21,
}]
- with pytest.raises((ValidationException,)):
+ with pytest.raises((ValidationErrors,)):
validation_raise(schema, *records)

Validation changes

Updates to validation to fit the comments from Scott. Validator Exceptions handled a bit better.