Files
deb-python-jsonschema/jsonschema.py
2013-01-12 22:32:18 -05:00

837 lines
25 KiB
Python

"""
An implementation of JSON Schema for Python
The main functionality is provided by the validator classes for each of the
supported JSON Schema versions.
Most commonly, :func:`validate` is the quickest way to simply validate a given
instance under a schema, and will create a validator for you.
"""
from __future__ import division, unicode_literals
import collections
import json
import itertools
import operator
import re
import sys
__version__ = "0.8.0-b1"
PY3 = sys.version_info[0] >= 3
if PY3:
basestring = unicode = str
iteritems = operator.methodcaller("items")
from urllib import parse as urlparse
from urllib.parse import unquote
from urllib.request import urlopen
else:
from itertools import izip as zip
iteritems = operator.methodcaller("iteritems")
from urllib import unquote
from urllib2 import urlopen
import urlparse
FLOAT_TOLERANCE = 10 ** -15
validators = {}
def validator(version):
"""
Register a validator for a ``version`` of the specification.
:argument str version: an identifier to use as the version's name
:returns: a class decorator to decorate the validator with the version
"""
def _validator(cls):
validators[version] = cls
return cls
return _validator
class UnknownType(Exception):
"""
An unknown type was given.
"""
class InvalidRef(Exception):
"""
An invalid reference was given.
"""
class SchemaError(Exception):
"""
The provided schema is malformed.
The same attributes exist for ``SchemaError``s as for ``ValidationError``s.
"""
def __init__(self, message, validator=None, path=()):
super(SchemaError, self).__init__(message)
self.message = message
self.path = list(path)
self.validator = validator
class ValidationError(Exception):
"""
The instance didn't properly validate with the provided schema.
Relevant attributes are:
* ``message`` : a human readable message explaining the error
* ``path`` : a list containing the path to the offending element (or []
if the error happened globally) in *reverse* order (i.e.
deepest index first).
"""
def __init__(self, message, validator=None, path=()):
# Any validator that recurses must append to the ValidationError's
# path (e.g., properties and items)
super(ValidationError, self).__init__(message)
self.message = message
self.path = list(path)
self.validator = validator
@validator("draft3")
class Draft3Validator(object):
"""
A validator for JSON Schema draft 3.
"""
DEFAULT_TYPES = {
"array" : list, "boolean" : bool, "integer" : int, "null" : type(None),
"number" : (int, float), "object" : dict, "string" : basestring,
}
def __init__(self, schema, types=(), resolver=None):
"""
Initialize a validator.
``schema`` should be a *valid* JSON Schema object already converted to
a native Python object (typically a dict via ``json.load``).
``types`` is a mapping (or iterable of 2-tuples) containing additional
types or alternate types to verify via the 'type' property. For
instance, the default types for the 'number' JSON Schema type are
``int`` and ``float``. To override this behavior (e.g. for also
allowing ``decimal.Decimal``), pass ``types={"number" : (int, float,
decimal.Decimal)} *including* the default types if so desired, which
are fairly obvious but can be accessed via the ``DEFAULT_TYPES``
attribute on this class if necessary.
``resolver`` is an object that knows how to resolve JSON Schema refs.
If not specified it is an instance of :class:`RefResolver`.
"""
self._types = dict(self.DEFAULT_TYPES)
self._types.update(types)
if resolver is None:
resolver = RefResolver.from_schema(schema)
self.resolver = resolver
self.schema = schema
def is_type(self, instance, type):
"""
Check if an ``instance`` is of the provided (JSON Schema) ``type``.
"""
if type == "any":
return True
elif type not in self._types:
raise UnknownType(type)
type = self._types[type]
# bool inherits from int, so ensure bools aren't reported as integers
if isinstance(instance, bool):
type = _flatten(type)
if int in type and bool not in type:
return False
return isinstance(instance, type)
def is_valid(self, instance, _schema=None):
"""
Check if the ``instance`` is valid under the current schema.
Returns a bool indicating whether validation succeeded.
"""
error = next(self.iter_errors(instance, _schema), None)
return error is None
@classmethod
def check_schema(cls, schema):
"""
Validate a ``schema`` against the meta-schema to see if it is valid.
"""
for error in cls(cls.META_SCHEMA).iter_errors(schema):
raise SchemaError(
error.message, validator=error.validator, path=error.path,
)
def iter_errors(self, instance, _schema=None):
"""
Lazily yield each of the errors in the given ``instance``.
>>> schema = {
... "type" : "array",
... "items" : {"enum" : [1, 2, 3]},
... "maxItems" : 2,
... }
>>> v = Draft3Validator(schema)
>>> for error in sorted(v.iter_errors([2, 3, 4]), key=str):
... print(error)
4 is not one of [1, 2, 3]
[2, 3, 4] is too long
"""
if _schema is None:
_schema = self.schema
for k, v in iteritems(_schema):
validator = getattr(self, "validate_%s" % (k.lstrip("$"),), None)
if validator is None:
continue
errors = validator(v, instance, _schema) or ()
for error in errors:
# set the validator if it wasn't already set by the called fn
if error.validator is None:
error.validator = k
yield error
def validate(self, *args, **kwargs):
"""
Validate an ``instance`` under the given ``schema``.
"""
for error in self.iter_errors(*args, **kwargs):
raise error
def validate_type(self, types, instance, schema):
types = _list(types)
for type in types:
if type == "any" or (
# Ouch. Brain hurts. Two paths here, either we have a schema, then
# check if the instance is valid under it
self.is_type(type, "object") and self.is_valid(instance, type)
# Or we have a type as a string, just check if the instance is that
# type. Also, HACK: we can reach the `or` here if skip_types is
# something other than error. If so, bail out.
) or (
self.is_type(type, "string") and
(self.is_type(instance, type) or type not in self._types)
):
return
else:
yield ValidationError(_types_msg(instance, types))
def validate_properties(self, properties, instance, schema):
if not self.is_type(instance, "object"):
return
for property, subschema in iteritems(properties):
if property in instance:
for error in self.iter_errors(instance[property], subschema):
error.path.append(property)
yield error
elif subschema.get("required", False):
yield ValidationError(
"%r is a required property" % (property,),
validator="required",
path=[property],
)
def validate_patternProperties(self, patternProperties, instance, schema):
if not self.is_type(instance, "object"):
return
for pattern, subschema in iteritems(patternProperties):
for k, v in iteritems(instance):
if re.match(pattern, k):
for error in self.iter_errors(v, subschema):
yield error
def validate_additionalProperties(self, aP, instance, schema):
if not self.is_type(instance, "object"):
return
extras = set(_find_additional_properties(instance, schema))
if self.is_type(aP, "object"):
for extra in extras:
for error in self.iter_errors(instance[extra], aP):
yield error
elif not aP and extras:
error = "Additional properties are not allowed (%s %s unexpected)"
yield ValidationError(error % _extras_msg(extras))
def validate_dependencies(self, dependencies, instance, schema):
if not self.is_type(instance, "object"):
return
for property, dependency in iteritems(dependencies):
if property not in instance:
continue
if self.is_type(dependency, "object"):
for error in self.iter_errors(instance, dependency):
yield error
else:
dependencies = _list(dependency)
for dependency in dependencies:
if dependency not in instance:
yield ValidationError(
"%r is a dependency of %r" % (dependency, property)
)
def validate_items(self, items, instance, schema):
if not self.is_type(instance, "array"):
return
if self.is_type(items, "object"):
for index, item in enumerate(instance):
for error in self.iter_errors(item, items):
error.path.append(index)
yield error
else:
for (index, item), subschema in zip(enumerate(instance), items):
for error in self.iter_errors(item, subschema):
error.path.append(index)
yield error
def validate_additionalItems(self, aI, instance, schema):
if not self.is_type(instance, "array"):
return
if not self.is_type(schema.get("items"), "array"):
return
if self.is_type(aI, "object"):
for item in instance[len(schema):]:
for error in self.iter_errors(item, aI):
yield error
elif not aI and len(instance) > len(schema.get("items", [])):
error = "Additional items are not allowed (%s %s unexpected)"
yield ValidationError(
error % _extras_msg(instance[len(schema.get("items", [])):])
)
def validate_minimum(self, minimum, instance, schema):
if not self.is_type(instance, "number"):
return
instance = float(instance)
if schema.get("exclusiveMinimum", False):
failed = instance <= minimum
cmp = "less than or equal to"
else:
failed = instance < minimum
cmp = "less than"
if failed:
yield ValidationError(
"%r is %s the minimum of %r" % (instance, cmp, minimum)
)
def validate_maximum(self, maximum, instance, schema):
if not self.is_type(instance, "number"):
return
instance = float(instance)
if schema.get("exclusiveMaximum", False):
failed = instance >= maximum
cmp = "greater than or equal to"
else:
failed = instance > maximum
cmp = "greater than"
if failed:
yield ValidationError(
"%r is %s the maximum of %r" % (instance, cmp, maximum)
)
def validate_minItems(self, mI, instance, schema):
if self.is_type(instance, "array") and len(instance) < mI:
yield ValidationError("%r is too short" % (instance,))
def validate_maxItems(self, mI, instance, schema):
if self.is_type(instance, "array") and len(instance) > mI:
yield ValidationError("%r is too long" % (instance,))
def validate_uniqueItems(self, uI, instance, schema):
if uI and self.is_type(instance, "array") and not _uniq(instance):
yield ValidationError("%r has non-unique elements" % instance)
def validate_pattern(self, patrn, instance, schema):
if self.is_type(instance, "string") and not re.match(patrn, instance):
yield ValidationError("%r does not match %r" % (instance, patrn))
def validate_minLength(self, mL, instance, schema):
if self.is_type(instance, "string") and len(instance) < mL:
yield ValidationError("%r is too short" % (instance,))
def validate_maxLength(self, mL, instance, schema):
if self.is_type(instance, "string") and len(instance) > mL:
yield ValidationError("%r is too long" % (instance,))
def validate_enum(self, enums, instance, schema):
if instance not in enums:
yield ValidationError("%r is not one of %r" % (instance, enums))
def validate_divisibleBy(self, dB, instance, schema):
if not self.is_type(instance, "number"):
return
if isinstance(dB, float):
mod = instance % dB
failed = (mod > FLOAT_TOLERANCE) and (dB - mod) > FLOAT_TOLERANCE
else:
failed = instance % dB
if failed:
yield ValidationError("%r is not divisible by %r" % (instance, dB))
def validate_disallow(self, disallow, instance, schema):
for disallowed in _list(disallow):
if self.is_valid(instance, {"type" : [disallowed]}):
yield ValidationError(
"%r is disallowed for %r" % (disallowed, instance)
)
def validate_extends(self, extends, instance, schema):
if self.is_type(extends, "object"):
extends = [extends]
for subschema in extends:
for error in self.iter_errors(instance, subschema):
yield error
def validate_ref(self, ref, instance, schema):
resolved = self.resolver.resolve(ref)
for error in self.iter_errors(instance, resolved):
yield error
Draft3Validator.META_SCHEMA = {
"$schema" : "http://json-schema.org/draft-03/schema#",
"id" : "http://json-schema.org/draft-03/schema#",
"type" : "object",
"properties" : {
"type" : {
"type" : ["string", "array"],
"items" : {"type" : ["string", {"$ref" : "#"}]},
"uniqueItems" : True,
"default" : "any"
},
"properties" : {
"type" : "object",
"additionalProperties" : {"$ref" : "#", "type": "object"},
"default" : {}
},
"patternProperties" : {
"type" : "object",
"additionalProperties" : {"$ref" : "#"},
"default" : {}
},
"additionalProperties" : {
"type" : [{"$ref" : "#"}, "boolean"], "default" : {}
},
"items" : {
"type" : [{"$ref" : "#"}, "array"],
"items" : {"$ref" : "#"},
"default" : {}
},
"additionalItems" : {
"type" : [{"$ref" : "#"}, "boolean"], "default" : {}
},
"required" : {"type" : "boolean", "default" : False},
"dependencies" : {
"type" : ["string", "array", "object"],
"additionalProperties" : {
"type" : ["string", "array", {"$ref" : "#"}],
"items" : {"type" : "string"}
},
"default" : {}
},
"minimum" : {"type" : "number"},
"maximum" : {"type" : "number"},
"exclusiveMinimum" : {"type" : "boolean", "default" : False},
"exclusiveMaximum" : {"type" : "boolean", "default" : False},
"minItems" : {"type" : "integer", "minimum" : 0, "default" : 0},
"maxItems" : {"type" : "integer", "minimum" : 0},
"uniqueItems" : {"type" : "boolean", "default" : False},
"pattern" : {"type" : "string", "format" : "regex"},
"minLength" : {"type" : "integer", "minimum" : 0, "default" : 0},
"maxLength" : {"type" : "integer"},
"enum" : {"type" : "array", "minItems" : 1, "uniqueItems" : True},
"default" : {"type" : "any"},
"title" : {"type" : "string"},
"description" : {"type" : "string"},
"format" : {"type" : "string"},
"maxDecimal" : {"type" : "number", "minimum" : 0},
"divisibleBy" : {
"type" : "number",
"minimum" : 0,
"exclusiveMinimum" : True,
"default" : 1
},
"disallow" : {
"type" : ["string", "array"],
"items" : {"type" : ["string", {"$ref" : "#"}]},
"uniqueItems" : True
},
"extends" : {
"type" : [{"$ref" : "#"}, "array"],
"items" : {"$ref" : "#"},
"default" : {}
},
"id" : {"type" : "string", "format" : "uri"},
"$ref" : {"type" : "string", "format" : "uri"},
"$schema" : {"type" : "string", "format" : "uri"},
},
"dependencies" : {
"exclusiveMinimum" : "minimum", "exclusiveMaximum" : "maximum"
},
}
class RefResolver(object):
"""
Resolve JSON References.
:argument str base_uri: URI of the referring document
:argument referrer: the actual referring document
:argument dict store: a mapping from URIs to documents to cache
"""
def __init__(self, base_uri, referrer, store=()):
self.base_uri = base_uri
self.referrer = referrer
self.store = dict(store, **_meta_schemas())
@classmethod
def from_schema(cls, schema, *args, **kwargs):
"""
Construct a resolver from a JSON schema object.
:argument schema schema: the referring schema
:rtype: :class:`RefResolver`
"""
return cls(schema.get("id", ""), schema, *args, **kwargs)
def resolve(self, ref):
"""
Resolve a JSON ``ref``.
:argument str ref: reference to resolve
:returns: the referrant document
"""
base_uri = self.base_uri
uri, fragment = urlparse.urldefrag(urlparse.urljoin(base_uri, ref))
if uri in self.store:
document = self.store[uri]
elif not uri or uri == self.base_uri:
document = self.referrer
else:
document = self.resolve_remote(uri)
return self.resolve_fragment(document, fragment.lstrip("/"))
def resolve_fragment(self, document, fragment):
"""
Resolve a ``fragment`` within the referenced ``document``.
:argument document: the referrant document
:argument str fragment: a URI fragment to resolve within it
"""
parts = unquote(fragment).split("/") if fragment else []
for part in parts:
part = part.replace("~1", "/").replace("~0", "~")
if part not in document:
raise InvalidRef("Unresolvable JSON pointer: %r" % fragment)
document = document[part]
return document
def resolve_remote(self, uri):
"""
Resolve a remote ``uri``.
Does not check the store first.
:argument str uri: the URI to resolve
:returns: the retrieved document
"""
return json.load(urlopen(uri))
class ErrorTree(object):
"""
ErrorTrees make it easier to check which validations failed.
"""
def __init__(self, errors=()):
self.errors = {}
self._contents = collections.defaultdict(self.__class__)
for error in errors:
container = self
for element in reversed(error.path):
container = container[element]
container.errors[error.validator] = error
def __contains__(self, k):
return k in self._contents
def __getitem__(self, k):
"""
Retrieve the child tree with key ``k``.
"""
return self._contents[k]
def __setitem__(self, k, v):
self._contents[k] = v
def __iter__(self):
return iter(self._contents)
def __len__(self):
return self.total_errors
def __repr__(self):
return "<%s (%s total errors)>" % (self.__class__.__name__, len(self))
@property
def total_errors(self):
"""
The total number of errors in the entire tree, including children.
"""
child_errors = sum(len(tree) for _, tree in iteritems(self._contents))
return len(self.errors) + child_errors
def _meta_schemas():
"""
Collect the urls and meta schemas from each known validator.
"""
meta_schemas = (v.META_SCHEMA for v in validators.values())
return dict((urlparse.urldefrag(m["id"])[0], m) for m in meta_schemas)
def _find_additional_properties(instance, schema):
"""
Return the set of additional properties for the given ``instance``.
Weeds out properties that should have been validated by ``properties`` and
/ or ``patternProperties``.
Assumes ``instance`` is dict-like already.
"""
properties = schema.get("properties", {})
patterns = "|".join(schema.get("patternProperties", {}))
for property in instance:
if property not in properties:
if patterns and re.search(patterns, property):
continue
yield property
def _extras_msg(extras):
"""
Create an error message for extra items or properties.
"""
if len(extras) == 1:
verb = "was"
else:
verb = "were"
return ", ".join(repr(extra) for extra in extras), verb
def _types_msg(instance, types):
"""
Create an error message for a failure to match the given types.
If the ``instance`` is an object and contains a ``name`` property, it will
be considered to be a description of that object and used as its type.
Otherwise the message is simply the reprs of the given ``types``.
"""
reprs = []
for type in types:
try:
reprs.append(repr(type["name"]))
except Exception:
reprs.append(repr(type))
return "%r is not of type %s" % (instance, ", ".join(reprs))
def _flatten(suitable_for_isinstance):
"""
isinstance() can accept a bunch of really annoying different types:
* a single type
* a tuple of types
* an arbitrary nested tree of tuples
Return a flattened tuple of the given argument.
"""
types = set()
if not isinstance(suitable_for_isinstance, tuple):
suitable_for_isinstance = (suitable_for_isinstance,)
for thing in suitable_for_isinstance:
if isinstance(thing, tuple):
types.update(_flatten(thing))
else:
types.add(thing)
return tuple(types)
def _list(thing):
"""
Wrap ``thing`` in a list if it's a single str.
Otherwise, return it unchanged.
"""
if isinstance(thing, basestring):
return [thing]
return thing
def _delist(thing):
"""
Unwrap ``thing`` to a single element if its a single str in a list.
Otherwise, return it unchanged.
"""
if (
isinstance(thing, list) and
len(thing) == 1
and isinstance(thing[0], basestring)
):
return thing[0]
return thing
def _unbool(element, true=object(), false=object()):
"""
A hack to make True and 1 and False and 0 unique for _uniq.
"""
if element is True:
return true
elif element is False:
return false
return element
def _uniq(container):
"""
Check if all of a container's elements are unique.
Successively tries first to rely that the elements are hashable, then
falls back on them being sortable, and finally falls back on brute
force.
"""
try:
return len(set(_unbool(i) for i in container)) == len(container)
except TypeError:
try:
sort = sorted(_unbool(i) for i in container)
sliced = itertools.islice(sort, 1, None)
for i, j in zip(sort, sliced):
if i == j:
return False
except (NotImplementedError, TypeError):
seen = []
for e in container:
e = _unbool(e)
if e in seen:
return False
seen.append(e)
return True
def validate(instance, schema, cls=Draft3Validator, *args, **kwargs):
"""
Validate an ``instance`` under the given ``schema``.
First verifies that the provided schema is itself valid, since not doing so
can lead to less obvious failures when validating. If you know it is or
don't care, use ``YourValidator(schema).validate(instance)`` directly
instead (e.g. ``Draft3Validator``).
``cls`` is a validator class that will be used to validate the instance.
By default this is a draft 3 validator. Any other provided positional and
keyword arguments will be provided to this class when constructing a
validator.
"""
cls.check_schema(schema)
cls(schema, *args, **kwargs).validate(instance)