from __future__ import division import contextlib import json import numbers try: import requests except ImportError: requests = None from jsonschema import _utils, _validators from jsonschema.compat import ( Sequence, urljoin, urlsplit, urldefrag, unquote, urlopen, str_types, int_types, iteritems, lru_cache, ) from jsonschema.exceptions import ErrorTree # Backwards compatibility # noqa from jsonschema.exceptions import RefResolutionError, SchemaError, UnknownType _unset = _utils.Unset() validators = {} meta_schemas = _utils.URIDict() def validates(version): """ Register the decorated validator for a ``version`` of the specification. Registered validators and their meta schemas will be considered when parsing ``$schema`` properties' URIs. :argument str version: an identifier to use as the version's name :returns: a class decorator to decorate the validator with the version """ def _validates(cls): validators[version] = cls if u"id" in cls.META_SCHEMA: meta_schemas[cls.META_SCHEMA[u"id"]] = cls return cls return _validates def create(meta_schema, validators=(), version=None, default_types=None): # noqa if default_types is None: default_types = { u"array" : list, u"boolean" : bool, u"integer" : int_types, u"null" : type(None), u"number" : numbers.Number, u"object" : dict, u"string" : str_types, } class Validator(object): VALIDATORS = dict(validators) META_SCHEMA = dict(meta_schema) DEFAULT_TYPES = dict(default_types) def __init__( self, schema, types=(), resolver=None, format_checker=None, ): self._types = dict(self.DEFAULT_TYPES) self._types.update(types) if resolver is None: resolver = RefResolver.from_schema(schema) self.resolver = resolver self.format_checker = format_checker self.schema = schema @classmethod def check_schema(cls, schema): for error in cls(cls.META_SCHEMA).iter_errors(schema): raise SchemaError.create_from(error) def iter_errors(self, instance, _schema=None): if _schema is None: _schema = self.schema scope = _schema.get(u"id") if scope: self.resolver.push_scope(scope) try: ref = _schema.get(u"$ref") if ref is not None: validators = [(u"$ref", ref)] else: validators = iteritems(_schema) for k, v in validators: validator = self.VALIDATORS.get(k) if validator is None: continue errors = validator(self, v, instance, _schema) or () for error in errors: # set details if not already set by the called fn error._set( validator=k, validator_value=v, instance=instance, schema=_schema, ) if k != u"$ref": error.schema_path.appendleft(k) yield error finally: if scope: self.resolver.pop_scope() def descend(self, instance, schema, path=None, schema_path=None): for error in self.iter_errors(instance, schema): if path is not None: error.path.appendleft(path) if schema_path is not None: error.schema_path.appendleft(schema_path) yield error def validate(self, *args, **kwargs): for error in self.iter_errors(*args, **kwargs): raise error def is_type(self, instance, type): if type not in self._types: raise UnknownType(type, instance, self.schema) pytypes = self._types[type] # bool inherits from int, so ensure bools aren't reported as ints if isinstance(instance, bool): pytypes = _utils.flatten(pytypes) is_number = any( issubclass(pytype, numbers.Number) for pytype in pytypes ) if is_number and bool not in pytypes: return False return isinstance(instance, pytypes) def is_valid(self, instance, _schema=None): error = next(self.iter_errors(instance, _schema), None) return error is None if version is not None: Validator = validates(version)(Validator) Validator.__name__ = version.title().replace(" ", "") + "Validator" return Validator def extend(validator, validators, version=None): all_validators = dict(validator.VALIDATORS) all_validators.update(validators) return create( meta_schema=validator.META_SCHEMA, validators=all_validators, version=version, default_types=validator.DEFAULT_TYPES, ) Draft3Validator = create( meta_schema=_utils.load_schema("draft3"), validators={ u"$ref" : _validators.ref, u"additionalItems" : _validators.additionalItems, u"additionalProperties" : _validators.additionalProperties, u"dependencies" : _validators.dependencies, u"disallow" : _validators.disallow_draft3, u"divisibleBy" : _validators.multipleOf, u"enum" : _validators.enum, u"extends" : _validators.extends_draft3, u"format" : _validators.format, u"items" : _validators.items, u"maxItems" : _validators.maxItems, u"maxLength" : _validators.maxLength, u"maximum" : _validators.maximum, u"minItems" : _validators.minItems, u"minLength" : _validators.minLength, u"minimum" : _validators.minimum, u"multipleOf" : _validators.multipleOf, u"pattern" : _validators.pattern, u"patternProperties" : _validators.patternProperties, u"properties" : _validators.properties_draft3, u"type" : _validators.type_draft3, u"uniqueItems" : _validators.uniqueItems, }, version="draft3", ) Draft4Validator = create( meta_schema=_utils.load_schema("draft4"), validators={ u"$ref" : _validators.ref, u"additionalItems" : _validators.additionalItems, u"additionalProperties" : _validators.additionalProperties, u"allOf" : _validators.allOf_draft4, u"anyOf" : _validators.anyOf_draft4, u"dependencies" : _validators.dependencies, u"enum" : _validators.enum, u"format" : _validators.format, u"items" : _validators.items, u"maxItems" : _validators.maxItems, u"maxLength" : _validators.maxLength, u"maxProperties" : _validators.maxProperties_draft4, u"maximum" : _validators.maximum, u"minItems" : _validators.minItems, u"minLength" : _validators.minLength, u"minProperties" : _validators.minProperties_draft4, u"minimum" : _validators.minimum, u"multipleOf" : _validators.multipleOf, u"not" : _validators.not_draft4, u"oneOf" : _validators.oneOf_draft4, u"pattern" : _validators.pattern, u"patternProperties" : _validators.patternProperties, u"properties" : _validators.properties_draft4, u"required" : _validators.required_draft4, u"type" : _validators.type_draft4, u"uniqueItems" : _validators.uniqueItems, }, version="draft4", ) 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 :argument bool cache_remote: whether remote refs should be cached after first resolution :argument dict handlers: a mapping from URI schemes to functions that should be used to retrieve them :arguments functools.lru_cache urljoin_cache: a cache that will be used for caching the results of joining the resolution scope to subscopes. :arguments functools.lru_cache remote_cache: a cache that will be used for caching the results of resolved remote URLs. """ def __init__( self, base_uri, referrer, store=(), cache_remote=True, handlers=(), urljoin_cache=None, remote_cache=None, ): if urljoin_cache is None: urljoin_cache = lru_cache(1024)(urljoin) if remote_cache is None: remote_cache = lru_cache(1024)(self.resolve_from_url) self.referrer = referrer self.cache_remote = cache_remote self.handlers = dict(handlers) self._scopes_stack = [base_uri] self.store = _utils.URIDict( (id, validator.META_SCHEMA) for id, validator in iteritems(meta_schemas) ) self.store.update(store) self.store[base_uri] = referrer self._urljoin_cache = urljoin_cache self._remote_cache = remote_cache @classmethod def from_schema(cls, schema, *args, **kwargs): """ Construct a resolver from a JSON schema object. :argument schema: the referring schema :rtype: :class:`RefResolver` """ return cls(schema.get(u"id", u""), schema, *args, **kwargs) def push_scope(self, scope): self._scopes_stack.append( self._urljoin_cache(self.resolution_scope, scope), ) def pop_scope(self): try: self._scopes_stack.pop() except IndexError: raise RefResolutionError( "Failed to pop the scope from an empty stack. " "`pop_scope()` should only be called once for every " "`push_scope()`", ) @property def resolution_scope(self): return self._scopes_stack[-1] @property def base_uri(self): uri, _ = urldefrag(self.resolution_scope) return uri @contextlib.contextmanager def in_scope(self, scope): self.push_scope(scope) try: yield finally: self.pop_scope() @contextlib.contextmanager def resolving(self, ref): """ Context manager which resolves a JSON ``ref`` and enters the resolution scope of this ref. :argument str ref: reference to resolve """ url, resolved = self.resolve(ref) self.push_scope(url) try: yield resolved finally: self.pop_scope() def resolve(self, ref): url = self._urljoin_cache(self.resolution_scope, ref) return url, self._remote_cache(url) def resolve_from_url(self, url): url, fragment = urldefrag(url) try: document = self.store[url] except KeyError: try: document = self.resolve_remote(url) except Exception as exc: raise RefResolutionError(exc) return self.resolve_fragment(document, fragment) 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 """ fragment = fragment.lstrip(u"/") parts = unquote(fragment).split(u"/") if fragment else [] for part in parts: part = part.replace(u"~1", u"/").replace(u"~0", u"~") if isinstance(document, Sequence): # Array indexes should be turned into integers try: part = int(part) except ValueError: pass try: document = document[part] except (TypeError, LookupError): raise RefResolutionError( "Unresolvable JSON pointer: %r" % fragment ) return document def resolve_remote(self, uri): """ Resolve a remote ``uri``. If called directly, does not check the store first, but after retrieving the document at the specified URI it will be saved in the store if :attr:`cache_remote` is True. .. note:: If the requests_ library is present, ``jsonschema`` will use it to request the remote ``uri``, so that the correct encoding is detected and used. If it isn't, or if the scheme of the ``uri`` is not ``http`` or ``https``, UTF-8 is assumed. :argument str uri: the URI to resolve :returns: the retrieved document .. _requests: http://pypi.python.org/pypi/requests/ """ scheme = urlsplit(uri).scheme if scheme in self.handlers: result = self.handlers[scheme](uri) elif ( scheme in [u"http", u"https"] and requests and getattr(requests.Response, "json", None) is not None ): # Requests has support for detecting the correct encoding of # json over http if callable(requests.Response.json): result = requests.get(uri).json() else: result = requests.get(uri).json else: # Otherwise, pass off to urllib and assume utf-8 result = json.loads(urlopen(uri).read().decode("utf-8")) if self.cache_remote: self.store[uri] = result return result def validator_for(schema, default=_unset): if default is _unset: default = Draft4Validator return meta_schemas.get(schema.get(u"$schema", u""), default) def validate(instance, schema, cls=None, *args, **kwargs): """ Validate an instance under the given schema. >>> validate([2, 3, 4], {"maxItems" : 2}) Traceback (most recent call last): ... ValidationError: [2, 3, 4] is too long :func:`validate` will first verify that the provided schema is itself valid, since not doing so can lead to less obvious error messages and fail in less obvious or consistent ways. If you know you have a valid schema already or don't care, you might prefer using the :meth:`~IValidator.validate` method directly on a specific validator (e.g. :meth:`Draft4Validator.validate`). :argument instance: the instance to validate :argument schema: the schema to validate with :argument cls: an :class:`IValidator` class that will be used to validate the instance. If the ``cls`` argument is not provided, two things will happen in accordance with the specification. First, if the schema has a :validator:`$schema` property containing a known meta-schema [#]_ then the proper validator will be used. The specification recommends that all schemas contain :validator:`$schema` properties for this reason. If no :validator:`$schema` property is found, the default validator class is :class:`Draft4Validator`. Any other provided positional and keyword arguments will be passed on when instantiating the ``cls``. :raises: :exc:`ValidationError` if the instance is invalid :exc:`SchemaError` if the schema itself is invalid .. rubric:: Footnotes .. [#] known by a validator registered with :func:`validates` """ if cls is None: cls = validator_for(schema) cls.check_schema(schema) cls(schema, *args, **kwargs).validate(instance)