"""
Creation
and extension of validators,
with implementations
for existing drafts.
"""
from __future__
import annotations
from collections
import deque
from collections.abc
import Mapping, Sequence
from functools
import lru_cache
from operator
import methodcaller
from urllib.parse
import unquote, urldefrag, urljoin, urlsplit
from urllib.request
import urlopen
from warnings
import warn
import contextlib
import json
import reprlib
import typing
import warnings
from pyrsistent
import m
import attr
from jsonschema
import (
_format,
_legacy_validators,
_types,
_utils,
_validators,
exceptions,
)
_UNSET = _utils.Unset()
_VALIDATORS: dict[str, typing.Any] = {}
_META_SCHEMAS = _utils.URIDict()
_VOCABULARIES: list[tuple[str, typing.Any]] = []
def __getattr__(name):
if name ==
"ErrorTree":
warnings.warn(
"Importing ErrorTree from jsonschema.validators is deprecated. "
"Instead import it from jsonschema.exceptions.",
DeprecationWarning,
stacklevel=2,
)
from jsonschema.exceptions
import ErrorTree
return ErrorTree
elif name ==
"validators":
warnings.warn(
"Accessing jsonschema.validators.validators is deprecated. "
"Use jsonschema.validators.validator_for with a given schema.",
DeprecationWarning,
stacklevel=2,
)
return _VALIDATORS
elif name ==
"meta_schemas":
warnings.warn(
"Accessing jsonschema.validators.meta_schemas is deprecated. "
"Use jsonschema.validators.validator_for with a given schema.",
DeprecationWarning,
stacklevel=2,
)
return _META_SCHEMAS
raise AttributeError(f
"module {__name__} has no attribute {name}")
def validates(version):
"""
Register the decorated validator
for a ``version`` of the specification.
Registered validators
and their meta schemas will be considered when
parsing :kw:`$schema` keywords
' URIs.
Arguments:
version (str):
An identifier to use
as the version
's name
Returns:
collections.abc.Callable:
a
class decorator to decorate the validator
with the version
"""
def _validates(cls):
_VALIDATORS[version] = cls
meta_schema_id = cls.ID_OF(cls.META_SCHEMA)
_META_SCHEMAS[meta_schema_id] = cls
return cls
return _validates
def _id_of(schema):
"""
Return the ID of a schema
for recent JSON Schema drafts.
"""
if schema
is True or schema
is False:
return ""
return schema.get(
"$id",
"")
def _store_schema_list():
if not _VOCABULARIES:
package = _utils.resources.files(__package__)
for version
in package.joinpath(
"schemas",
"vocabularies").iterdir():
for path
in version.iterdir():
vocabulary = json.loads(path.read_text())
_VOCABULARIES.append((vocabulary[
"$id"], vocabulary))
return [
(id, validator.META_SCHEMA)
for id, validator
in _META_SCHEMAS.items()
] + _VOCABULARIES
def create(
meta_schema,
validators=(),
version=
None,
type_checker=_types.draft202012_type_checker,
format_checker=_format.draft202012_format_checker,
id_of=_id_of,
applicable_validators=methodcaller(
"items"),
):
"""
Create a new validator
class.
Arguments:
meta_schema (collections.abc.Mapping):
the meta schema
for the new validator
class
validators (collections.abc.Mapping):
a mapping
from names to callables, where each callable will
validate the schema property
with the given name.
Each callable should take 4 arguments:
1. a validator instance,
2. the value of the property being validated within the
instance
3. the instance
4. the schema
version (str):
an identifier
for the version that this validator
class will
validate.
If provided, the returned validator
class will
have its ``__name__`` set to include the version,
and also
will have `jsonschema.validators.validates` automatically
called
for the given version.
type_checker (jsonschema.TypeChecker):
a type checker, used when applying the :kw:`type` keyword.
If unprovided, a `jsonschema.TypeChecker` will be created
with a set of default types typical of JSON Schema drafts.
format_checker (jsonschema.FormatChecker):
a format checker, used when applying the :kw:`format` keyword.
If unprovided, a `jsonschema.FormatChecker` will be created
with a set of default formats typical of JSON Schema drafts.
id_of (collections.abc.Callable):
A function that given a schema, returns its ID.
applicable_validators (collections.abc.Callable):
A function that given a schema, returns the list of
applicable validators (validation keywords
and callables)
which will be used to validate the instance.
Returns:
a new `jsonschema.protocols.Validator`
class
"""
# preemptively don't shadow the `Validator.format_checker` local
format_checker_arg = format_checker
@attr.s
class Validator:
VALIDATORS = dict(validators)
META_SCHEMA = dict(meta_schema)
TYPE_CHECKER = type_checker
FORMAT_CHECKER = format_checker_arg
ID_OF = staticmethod(id_of)
schema = attr.ib(repr=reprlib.repr)
resolver = attr.ib(default=
None, repr=
False)
format_checker = attr.ib(default=
None)
def __init_subclass__(cls):
warnings.warn(
(
"Subclassing validator classes is not intended to "
"be part of their public API. A future version "
"will make doing so an error, as the behavior of "
"subclasses isn't guaranteed to stay the same "
"between releases of jsonschema. Instead, prefer "
"composition of validators, wrapping them in an object "
"owned entirely by the downstream library."
),
DeprecationWarning,
stacklevel=2,
)
def __attrs_post_init__(self):
if self.resolver
is None:
self.resolver = RefResolver.from_schema(
self.schema,
id_of=id_of,
)
@classmethod
def check_schema(cls, schema, format_checker=_UNSET):
Validator = validator_for(cls.META_SCHEMA, default=cls)
if format_checker
is _UNSET:
format_checker = Validator.FORMAT_CHECKER
validator = Validator(
schema=cls.META_SCHEMA,
format_checker=format_checker,
)
for error
in validator.iter_errors(schema):
raise exceptions.SchemaError.create_from(error)
def evolve(self, **changes):
# Essentially reproduces attr.evolve, but may involve instantiating
# a different class than this one.
cls = self.__class__
schema = changes.setdefault(
"schema", self.schema)
NewValidator = validator_for(schema, default=cls)
for field
in attr.fields(cls):
if not field.init:
continue
attr_name = field.name
# To deal with private attributes.
init_name = attr_name
if attr_name[0] !=
"_" else attr_name[1:]
if init_name
not in changes:
changes[init_name] = getattr(self, attr_name)
return NewValidator(**changes)
def iter_errors(self, instance, _schema=
None):
if _schema
is not None:
warnings.warn(
(
"Passing a schema to Validator.iter_errors "
"is deprecated and will be removed in a future "
"release. Call validator.evolve(schema=new_schema)."
"iter_errors(...) instead."
),
DeprecationWarning,
stacklevel=2,
)
else:
_schema = self.schema
if _schema
is True:
return
elif _schema
is False:
yield exceptions.ValidationError(
f
"False schema does not allow {instance!r}",
validator=
None,
validator_value=
None,
instance=instance,
schema=_schema,
)
return
scope = id_of(_schema)
if scope:
self.resolver.push_scope(scope)
try:
for k, v
in applicable_validators(_schema):
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,
type_checker=self.TYPE_CHECKER,
)
if k
not in {
"if",
"$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.evolve(schema=schema).iter_errors(instance):
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):
try:
return self.TYPE_CHECKER.is_type(instance, type)
except exceptions.UndefinedTypeCheck:
raise exceptions.UnknownType(type, instance, self.schema)
def is_valid(self, instance, _schema=
None):
if _schema
is not None:
warnings.warn(
(
"Passing a schema to Validator.is_valid is deprecated "
"and will be removed in a future release. Call "
"validator.evolve(schema=new_schema).is_valid(...) "
"instead."
),
DeprecationWarning,
stacklevel=2,
)
self = self.evolve(schema=_schema)
error = next(self.iter_errors(instance),
None)
return error
is None
if version
is not None:
safe = version.title().replace(
" ",
"").replace(
"-",
"")
Validator.__name__ = Validator.__qualname__ = f
"{safe}Validator"
Validator = validates(version)(Validator)
return Validator
def extend(
validator,
validators=(),
version=
None,
type_checker=
None,
format_checker=
None,
):
"""
Create a new validator
class by extending an existing one.
Arguments:
validator (jsonschema.protocols.Validator):
an existing validator
class
validators (collections.abc.Mapping):
a mapping of new validator callables to extend
with, whose
structure
is as in `create`.
.. note::
Any validator callables
with the same name
as an
existing one will (silently) replace the old validator
callable entirely, effectively overriding any validation
done
in the
"parent" validator
class.
If you wish to instead extend the behavior of a parent
's
validator callable, delegate
and call it directly
in
the new validator function by retrieving it using
``OldValidator.VALIDATORS[
"validation_keyword_name"]``.
version (str):
a version
for the new validator
class
type_checker (jsonschema.TypeChecker):
a type checker, used when applying the :kw:`type` keyword.
If unprovided, the type checker of the extended
`jsonschema.protocols.Validator` will be carried along.
format_checker (jsonschema.FormatChecker):
a format checker, used when applying the :kw:`format` keyword.
If unprovided, the format checker of the extended
`jsonschema.protocols.Validator` will be carried along.
Returns:
a new `jsonschema.protocols.Validator`
class extending the one
provided
.. note:: Meta Schemas
The new validator
class will have its parent
's meta schema.
If you wish to change
or extend the meta schema
in the new
validator
class, modify ``META_SCHEMA`` directly on the returned
class. Note that no implicit copying
is done, so a copy should
likely be made before modifying it,
in order to
not affect the
old validator.
"""
all_validators = dict(validator.VALIDATORS)
all_validators.update(validators)
if type_checker
is None:
type_checker = validator.TYPE_CHECKER
if format_checker
is None:
format_checker = validator.FORMAT_CHECKER
return create(
meta_schema=validator.META_SCHEMA,
validators=all_validators,
version=version,
type_checker=type_checker,
format_checker=format_checker,
id_of=validator.ID_OF,
)
Draft3Validator = create(
meta_schema=_utils.load_schema(
"draft3"),
validators={
"$ref": _validators.ref,
"additionalItems": _validators.additionalItems,
"additionalProperties": _validators.additionalProperties,
"dependencies": _legacy_validators.dependencies_draft3,
"disallow": _legacy_validators.disallow_draft3,
"divisibleBy": _validators.multipleOf,
"enum": _validators.enum,
"extends": _legacy_validators.extends_draft3,
"format": _validators.format,
"items": _legacy_validators.items_draft3_draft4,
"maxItems": _validators.maxItems,
"maxLength": _validators.maxLength,
"maximum": _legacy_validators.maximum_draft3_draft4,
"minItems": _validators.minItems,
"minLength": _validators.minLength,
"minimum": _legacy_validators.minimum_draft3_draft4,
"pattern": _validators.pattern,
"patternProperties": _validators.patternProperties,
"properties": _legacy_validators.properties_draft3,
"type": _legacy_validators.type_draft3,
"uniqueItems": _validators.uniqueItems,
},
type_checker=_types.draft3_type_checker,
format_checker=_format.draft3_format_checker,
version=
"draft3",
id_of=_legacy_validators.id_of_ignore_ref(property=
"id"),
applicable_validators=_legacy_validators.ignore_ref_siblings,
)
Draft4Validator = create(
meta_schema=_utils.load_schema(
"draft4"),
validators={
"$ref": _validators.ref,
"additionalItems": _validators.additionalItems,
"additionalProperties": _validators.additionalProperties,
"allOf": _validators.allOf,
"anyOf": _validators.anyOf,
"dependencies": _legacy_validators.dependencies_draft4_draft6_draft7,
"enum": _validators.enum,
"format": _validators.format,
"items": _legacy_validators.items_draft3_draft4,
"maxItems": _validators.maxItems,
"maxLength": _validators.maxLength,
"maxProperties": _validators.maxProperties,
"maximum": _legacy_validators.maximum_draft3_draft4,
"minItems": _validators.minItems,
"minLength": _validators.minLength,
"minProperties": _validators.minProperties,
"minimum": _legacy_validators.minimum_draft3_draft4,
"multipleOf": _validators.multipleOf,
"not": _validators.not_,
"oneOf": _validators.oneOf,
"pattern": _validators.pattern,
"patternProperties": _validators.patternProperties,
"properties": _validators.properties,
"required": _validators.required,
"type": _validators.type,
"uniqueItems": _validators.uniqueItems,
},
type_checker=_types.draft4_type_checker,
format_checker=_format.draft4_format_checker,
version=
"draft4",
id_of=_legacy_validators.id_of_ignore_ref(property=
"id"),
applicable_validators=_legacy_validators.ignore_ref_siblings,
)
Draft6Validator = create(
meta_schema=_utils.load_schema(
"draft6"),
validators={
"$ref": _validators.ref,
"additionalItems": _validators.additionalItems,
"additionalProperties": _validators.additionalProperties,
"allOf": _validators.allOf,
"anyOf": _validators.anyOf,
"const": _validators.const,
"contains": _legacy_validators.contains_draft6_draft7,
"dependencies": _legacy_validators.dependencies_draft4_draft6_draft7,
"enum": _validators.enum,
"exclusiveMaximum": _validators.exclusiveMaximum,
"exclusiveMinimum": _validators.exclusiveMinimum,
"format": _validators.format,
"items": _legacy_validators.items_draft6_draft7_draft201909,
"maxItems": _validators.maxItems,
"maxLength": _validators.maxLength,
"maxProperties": _validators.maxProperties,
"maximum": _validators.maximum,
"minItems": _validators.minItems,
"minLength": _validators.minLength,
"minProperties": _validators.minProperties,
"minimum": _validators.minimum,
"multipleOf": _validators.multipleOf,
"not": _validators.not_,
"oneOf": _validators.oneOf,
"pattern": _validators.pattern,
"patternProperties": _validators.patternProperties,
"properties": _validators.properties,
"propertyNames": _validators.propertyNames,
"required": _validators.required,
"type": _validators.type,
"uniqueItems": _validators.uniqueItems,
},
type_checker=_types.draft6_type_checker,
format_checker=_format.draft6_format_checker,
version=
"draft6",
id_of=_legacy_validators.id_of_ignore_ref(),
applicable_validators=_legacy_validators.ignore_ref_siblings,
)
Draft7Validator = create(
meta_schema=_utils.load_schema(
"draft7"),
validators={
"$ref": _validators.ref,
"additionalItems": _validators.additionalItems,
"additionalProperties": _validators.additionalProperties,
"allOf": _validators.allOf,
"anyOf": _validators.anyOf,
"const": _validators.const,
"contains": _legacy_validators.contains_draft6_draft7,
"dependencies": _legacy_validators.dependencies_draft4_draft6_draft7,
"enum": _validators.enum,
"exclusiveMaximum": _validators.exclusiveMaximum,
"exclusiveMinimum": _validators.exclusiveMinimum,
"format": _validators.format,
"if": _validators.if_,
"items": _legacy_validators.items_draft6_draft7_draft201909,
"maxItems": _validators.maxItems,
"maxLength": _validators.maxLength,
"maxProperties": _validators.maxProperties,
"maximum": _validators.maximum,
"minItems": _validators.minItems,
"minLength": _validators.minLength,
"minProperties": _validators.minProperties,
"minimum": _validators.minimum,
"multipleOf": _validators.multipleOf,
"not": _validators.not_,
"oneOf": _validators.oneOf,
"pattern": _validators.pattern,
"patternProperties": _validators.patternProperties,
"properties": _validators.properties,
"propertyNames": _validators.propertyNames,
"required": _validators.required,
"type": _validators.type,
"uniqueItems": _validators.uniqueItems,
},
type_checker=_types.draft7_type_checker,
format_checker=_format.draft7_format_checker,
version=
"draft7",
id_of=_legacy_validators.id_of_ignore_ref(),
applicable_validators=_legacy_validators.ignore_ref_siblings,
)
Draft201909Validator = create(
meta_schema=_utils.load_schema(
"draft2019-09"),
validators={
"$recursiveRef": _legacy_validators.recursiveRef,
"$ref": _validators.ref,
"additionalItems": _validators.additionalItems,
"additionalProperties": _validators.additionalProperties,
"allOf": _validators.allOf,
"anyOf": _validators.anyOf,
"const": _validators.const,
"contains": _validators.contains,
"dependentRequired": _validators.dependentRequired,
"dependentSchemas": _validators.dependentSchemas,
"enum": _validators.enum,
"exclusiveMaximum": _validators.exclusiveMaximum,
"exclusiveMinimum": _validators.exclusiveMinimum,
"format": _validators.format,
"if": _validators.if_,
"items": _legacy_validators.items_draft6_draft7_draft201909,
"maxItems": _validators.maxItems,
"maxLength": _validators.maxLength,
"maxProperties": _validators.maxProperties,
"maximum": _validators.maximum,
"minItems": _validators.minItems,
"minLength": _validators.minLength,
"minProperties": _validators.minProperties,
"minimum": _validators.minimum,
"multipleOf": _validators.multipleOf,
"not": _validators.not_,
"oneOf": _validators.oneOf,
"pattern": _validators.pattern,
"patternProperties": _validators.patternProperties,
"properties": _validators.properties,
"propertyNames": _validators.propertyNames,
"required": _validators.required,
"type": _validators.type,
"unevaluatedItems": _legacy_validators.unevaluatedItems_draft2019,
"unevaluatedProperties": _validators.unevaluatedProperties,
"uniqueItems": _validators.uniqueItems,
},
type_checker=_types.draft201909_type_checker,
format_checker=_format.draft201909_format_checker,
version=
"draft2019-09",
)
Draft202012Validator = create(
meta_schema=_utils.load_schema(
"draft2020-12"),
validators={
"$dynamicRef": _validators.dynamicRef,
"$ref": _validators.ref,
"additionalItems": _validators.additionalItems,
"additionalProperties": _validators.additionalProperties,
"allOf": _validators.allOf,
"anyOf": _validators.anyOf,
"const": _validators.const,
"contains": _validators.contains,
"dependentRequired": _validators.dependentRequired,
"dependentSchemas": _validators.dependentSchemas,
"enum": _validators.enum,
"exclusiveMaximum": _validators.exclusiveMaximum,
"exclusiveMinimum": _validators.exclusiveMinimum,
"format": _validators.format,
"if": _validators.if_,
"items": _validators.items,
"maxItems": _validators.maxItems,
"maxLength": _validators.maxLength,
"maxProperties": _validators.maxProperties,
"maximum": _validators.maximum,
"minItems": _validators.minItems,
"minLength": _validators.minLength,
"minProperties": _validators.minProperties,
"minimum": _validators.minimum,
"multipleOf": _validators.multipleOf,
"not": _validators.not_,
"oneOf": _validators.oneOf,
"pattern": _validators.pattern,
"patternProperties": _validators.patternProperties,
"prefixItems": _validators.prefixItems,
"properties": _validators.properties,
"propertyNames": _validators.propertyNames,
"required": _validators.required,
"type": _validators.type,
"unevaluatedItems": _validators.unevaluatedItems,
"unevaluatedProperties": _validators.unevaluatedProperties,
"uniqueItems": _validators.uniqueItems,
},
type_checker=_types.draft202012_type_checker,
format_checker=_format.draft202012_format_checker,
version=
"draft2020-12",
)
_LATEST_VERSION = Draft202012Validator
class RefResolver:
"""
Resolve JSON References.
Arguments:
base_uri (str):
The URI of the referring document
referrer:
The actual referring document
store (dict):
A mapping
from URIs to documents to cache
cache_remote (bool):
Whether remote refs should be cached after first resolution
handlers (dict):
A mapping
from URI schemes to functions that should be used
to retrieve them
urljoin_cache (:func:`functools.lru_cache`):
A cache that will be used
for caching the results of joining
the resolution scope to subscopes.
remote_cache (:func:`functools.lru_cache`):
A cache that will be used
for caching the results of
resolved remote URLs.
Attributes:
cache_remote (bool):
Whether remote refs should be cached after first resolution
"""
def __init__(
self,
base_uri,
referrer,
store=m(),
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(_store_schema_list())
self.store.update(store)
self.store.update(
(schema[
"$id"], schema)
for schema
in store.values()
if isinstance(schema, Mapping)
and "$id" in schema
)
self.store[base_uri] = referrer
self._urljoin_cache = urljoin_cache
self._remote_cache = remote_cache
@classmethod
def from_schema(cls, schema, id_of=_id_of, *args, **kwargs):
"""
Construct a resolver
from a JSON schema object.
Arguments:
schema:
the referring schema
Returns:
`RefResolver`
"""
return cls(base_uri=id_of(schema), referrer=schema, *args, **kwargs)
# noqa: B026, E501
def push_scope(self, scope):
"""
Enter a given sub-scope.
Treats further dereferences
as being performed underneath the
given scope.
"""
self._scopes_stack.append(
self._urljoin_cache(self.resolution_scope, scope),
)
def pop_scope(self):
"""
Exit the most recent entered scope.
Treats further dereferences
as being performed underneath the
original scope.
Don
't call this method more times than `push_scope` has been
called.
"""
try:
self._scopes_stack.pop()
except IndexError:
raise exceptions.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):
"""
Retrieve the current resolution scope.
"""
return self._scopes_stack[-1]
@property
def base_uri(self):
"""
Retrieve the current base URI,
not including any fragment.
"""
uri, _ = urldefrag(self.resolution_scope)
return uri
@contextlib.contextmanager
def in_scope(self, scope):
"""
Temporarily enter the given scope
for the duration of the context.
.. deprecated:: v4.0.0
"""
warnings.warn(
"jsonschema.RefResolver.in_scope is deprecated and will be "
"removed in a future release.",
DeprecationWarning,
stacklevel=3,
)
self.push_scope(scope)
try:
yield
finally:
self.pop_scope()
@contextlib.contextmanager
def resolving(self, ref):
"""
Resolve the given ``ref``
and enter its resolution scope.
Exits the scope on exit of this context manager.
Arguments:
ref (str):
The reference to resolve
"""
url, resolved = self.resolve(ref)
self.push_scope(url)
try:
yield resolved
finally:
self.pop_scope()
def _find_in_referrer(self, key):
return self._get_subschemas_cache()[key]
@lru_cache()
# noqa: B019
def _get_subschemas_cache(self):
cache = {key: []
for key
in _SUBSCHEMAS_KEYWORDS}
for keyword, subschema
in _search_schema(
self.referrer, _match_subschema_keywords,
):
cache[keyword].append(subschema)
return cache
@lru_cache()
# noqa: B019
def _find_in_subschemas(self, url):
subschemas = self._get_subschemas_cache()[
"$id"]
if not subschemas:
return None
uri, fragment = urldefrag(url)
for subschema
in subschemas:
target_uri = self._urljoin_cache(
self.resolution_scope, subschema[
"$id"],
)
if target_uri.rstrip(
"/") == uri.rstrip(
"/"):
if fragment:
subschema = self.resolve_fragment(subschema, fragment)
self.store[url] = subschema
return url, subschema
return None
def resolve(self, ref):
"""
Resolve the given reference.
"""
url = self._urljoin_cache(self.resolution_scope, ref).rstrip(
"/")
match = self._find_in_subschemas(url)
if match
is not None:
return match
return url, self._remote_cache(url)
def resolve_from_url(self, url):
"""
Resolve the given URL.
"""
url, fragment = urldefrag(url)
if not url:
url = self.base_uri
try:
document = self.store[url]
except KeyError:
try:
document = self.resolve_remote(url)
except Exception
as exc:
raise exceptions.RefResolutionError(exc)
return self.resolve_fragment(document, fragment)
def resolve_fragment(self, document, fragment):
"""
Resolve a ``fragment`` within the referenced ``document``.
Arguments:
document:
The referent document
fragment (str):
a URI fragment to resolve within it
"""
fragment = fragment.lstrip(
"/")
if not fragment:
return document
if document
is self.referrer:
find = self._find_in_referrer
else:
def find(key):
yield from _search_schema(document, _match_keyword(key))
for keyword
in [
"$anchor",
"$dynamicAnchor"]:
for subschema
in find(keyword):
if fragment == subschema[keyword]:
return subschema
for keyword
in [
"id",
"$id"]:
for subschema
in find(keyword):
if "#" + fragment == subschema[keyword]:
return subschema
# Resolve via path
parts = unquote(fragment).split(
"/")
if fragment
else []
for part
in parts:
part = part.replace(
"~1",
"/").replace(
"~0",
"~")
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 exceptions.RefResolutionError(
f
"Unresolvable JSON pointer: {fragment!r}",
)
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.
Arguments:
uri (str):
The URI to resolve
Returns:
The retrieved document
.. _requests:
https://pypi.org/project/requests/
"""
try:
import requests
except ImportError:
requests =
None
scheme = urlsplit(uri).scheme
if scheme
in self.handlers:
result = self.handlers[scheme](uri)
elif scheme
in [
"http",
"https"]
and requests:
# Requests has support for detecting the correct encoding of
# json over http
result = requests.get(uri).json()
else:
# Otherwise, pass off to urllib and assume utf-8
with urlopen(uri)
as url:
result = json.loads(url.read().decode(
"utf-8"))
if self.cache_remote:
self.store[uri] = result
return result
_SUBSCHEMAS_KEYWORDS = (
"$id",
"id",
"$anchor",
"$dynamicAnchor")
def _match_keyword(keyword):
def matcher(value):
if keyword
in value:
yield value
return matcher
def _match_subschema_keywords(value):
for keyword
in _SUBSCHEMAS_KEYWORDS:
if keyword
in value:
yield keyword, value
def _search_schema(schema, matcher):
"""Breadth-first search routine."""
values = deque([schema])
while values:
value = values.pop()
if not isinstance(value, dict):
continue
yield from matcher(value)
values.extendleft(value.values())
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:`~jsonschema.validators.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, especially
if you intend to validate multiple instances
with
the same schema, you likely would prefer using the
`jsonschema.protocols.Validator.validate` method directly on a
specific validator (e.g. ``Draft20212Validator.validate``).
Arguments:
instance:
The instance to validate
schema:
The schema to validate
with
cls (jsonschema.protocols.Validator):
The
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
:kw:`$schema` keyword containing a known meta-schema [
#]_ then the
proper validator will be used. The specification recommends that
all schemas contain :kw:`$schema` properties
for this reason.
If no
:kw:`$schema` property
is found, the default validator
class is the
latest released draft.
Any other provided positional
and keyword arguments will be passed
on when instantiating the ``cls``.
Raises:
`jsonschema.exceptions.ValidationError`:
if the instance
is invalid
`jsonschema.exceptions.SchemaError`:
if the schema itself
is invalid
.. rubric:: Footnotes
.. [
#] known by a validator registered with
`jsonschema.validators.validates`
"""
if cls
is None:
cls = validator_for(schema)
cls.check_schema(schema)
validator = cls(schema, *args, **kwargs)
error = exceptions.best_match(validator.iter_errors(instance))
if error
is not None:
raise error
def validator_for(schema, default=_UNSET):
"""
Retrieve the validator
class appropriate
for validating the given schema.
Uses the :kw:`$schema` keyword that should be present
in the given
schema to look up the appropriate validator
class.
Arguments:
schema (collections.abc.Mapping
or bool):
the schema to look at
default:
the default to
return if the appropriate validator
class
cannot be determined.
If unprovided, the default
is to
return the latest supported
draft.
"""
DefaultValidator = _LATEST_VERSION
if default
is _UNSET
else default
if schema
is True or schema
is False or "$schema" not in schema:
return DefaultValidator
if schema[
"$schema"]
not in _META_SCHEMAS:
if default
is _UNSET:
warn(
(
"The metaschema specified by $schema was not found. "
"Using the latest draft to validate, but this will raise "
"an error in the future."
),
DeprecationWarning,
stacklevel=2,
)
return _META_SCHEMAS.get(schema[
"$schema"], DefaultValidator)
