Source code for jsonschema.validators

Creation and extension of validators, with implementations for existing drafts.
from __future__ import annotations

from collections import deque
from import Iterable, 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 warnings

from attrs import define, field, fields
from jsonschema_specifications import REGISTRY as SPECIFICATIONS
from rpds import HashTrieMap
import referencing.exceptions
import referencing.jsonschema

from jsonschema import (
from jsonschema.protocols import Validator

_UNSET = _utils.Unset()

_VALIDATORS: dict[str, Validator] = {}
_META_SCHEMAS = _utils.URIDict()

def __getattr__(name):
    if name == "ErrorTree":
            "Importing ErrorTree from jsonschema.validators is deprecated. "
            "Instead import it from jsonschema.exceptions.",
        from jsonschema.exceptions import ErrorTree
        return ErrorTree
    elif name == "validators":
            "Accessing jsonschema.validators.validators is deprecated. "
            "Use jsonschema.validators.validator_for with a given schema.",
        return _VALIDATORS
    elif name == "meta_schemas":
            "Accessing jsonschema.validators.meta_schemas is deprecated. "
            "Use jsonschema.validators.validator_for with a given schema.",
        return _META_SCHEMAS
    elif name == "RefResolver":
        return _RefResolver
    raise AttributeError(f"module {__name__} has no attribute {name}")

[docs] 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: 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 _warn_for_remote_retrieve(uri: str): from urllib.request import Request, urlopen headers = {"User-Agent": "python-jsonschema (deprecated $ref resolution)"} request = Request(uri, headers=headers) with urlopen(request) as response: warnings.warn( "Automatically retrieving remote references can be a security " "vulnerability and is discouraged by the JSON Schema " "specifications. Relying on this behavior is deprecated " "and will shortly become an error. If you are sure you want to " "remotely retrieve your reference and that it is safe to do so, " "you can find instructions for doing so via referencing.Registry " "in the referencing documentation " "(", DeprecationWarning, stacklevel=9, # Ha ha ha ha magic numbers :/ ) return referencing.Resource.from_contents( json.load(response), default_specification=referencing.jsonschema.DRAFT202012, ) _REMOTE_WARNING_REGISTRY = SPECIFICATIONS.combine( referencing.Registry(retrieve=_warn_for_remote_retrieve), # type: ignore[call-arg] # noqa: E501 )
[docs] def create( meta_schema: referencing.jsonschema.ObjectSchema, validators: ( Mapping[str, _typing.SchemaKeywordValidator] | Iterable[tuple[str, _typing.SchemaKeywordValidator]] ) = (), version: str | None = None, type_checker: _types.TypeChecker = _types.draft202012_type_checker, format_checker: _format.FormatChecker = _format.draft202012_format_checker, id_of: _typing.id_of = referencing.jsonschema.DRAFT202012.id_of, applicable_validators: _typing.ApplicableValidators = methodcaller( "items", ), ): """ Create a new validator class. Arguments: meta_schema: the meta schema for the new validator class validators: 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: 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: 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: 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: A function that given a schema, returns its ID. applicable_validators: A function that, given a schema, returns the list of applicable schema keywords and associated values which will be used to validate the instance. This is mostly used to support pre-draft 7 versions of JSON Schema which specified behavior around ignoring keywords if they were siblings of a ``$ref`` keyword. If you're not attempting to implement similar behavior, you can typically ignore this argument and leave it at its default. Returns: a new `jsonschema.protocols.Validator` class """ # preemptively don't shadow the `Validator.format_checker` local format_checker_arg = format_checker specification = referencing.jsonschema.specification_with( dialect_id=id_of(meta_schema) or "urn:unknown-dialect", default=referencing.Specification.OPAQUE, ) @define class Validator: VALIDATORS = dict(validators) # noqa: RUF012 META_SCHEMA = dict(meta_schema) # noqa: RUF012 TYPE_CHECKER = type_checker FORMAT_CHECKER = format_checker_arg ID_OF = staticmethod(id_of) _APPLICABLE_VALIDATORS = applicable_validators schema: referencing.jsonschema.Schema = field(repr=reprlib.repr) _ref_resolver = field(default=None, repr=False, alias="resolver") format_checker: _format.FormatChecker | None = field(default=None) # TODO: include new meta-schemas added at runtime _registry: referencing.jsonschema.SchemaRegistry = field( default=_REMOTE_WARNING_REGISTRY, kw_only=True, repr=False, ) _resolver = field( alias="_resolver", default=None, kw_only=True, repr=False, ) 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 evolve(self, **changes): cls = self.__class__ schema = changes.setdefault("schema", self.schema) NewValidator = validator_for(schema, default=cls) for field in fields(cls): # noqa: F402 if not field.init: continue attr_name = init_name = field.alias if init_name not in changes: changes[init_name] = getattr(self, attr_name) return NewValidator(**changes) cls.evolve = evolve def __attrs_post_init__(self): if self._resolver is None: registry = self._registry if registry is not _REMOTE_WARNING_REGISTRY: registry = SPECIFICATIONS.combine(registry) resource = specification.create_resource(self.schema) self._resolver = registry.resolver_with_root(resource) # REMOVEME: Legacy ref resolution state management. push_scope = getattr(self._ref_resolver, "push_scope", None) if push_scope is not None: id = id_of(self.schema) if id is not None: push_scope(id) @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) @property def resolver(self): warnings.warn( ( f"Accessing {self.__class__.__name__}.resolver is " "deprecated as of v4.18.0, in favor of the " " " "library, which provides more compliant referencing " "behavior as well as more flexible APIs for " "customization." ), DeprecationWarning, stacklevel=2, ) if self._ref_resolver is None: self._ref_resolver = _RefResolver.from_schema( self.schema, id_of=id_of, ) return self._ref_resolver def evolve(self, **changes): schema = changes.setdefault("schema", self.schema) NewValidator = validator_for(schema, default=self.__class__) for (attr_name, init_name) in evolve_fields: 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 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 def descend( self, instance, schema, path=None, schema_path=None, resolver=None, ): 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 if self._ref_resolver is not None: evolved = self.evolve(schema=schema) else: if resolver is None: resolver = self._resolver.in_subresource( specification.create_resource(schema), ) evolved = self.evolve(schema=schema, _resolver=resolver) for k, v in applicable_validators(schema): validator = evolved.VALIDATORS.get(k) if validator is None: continue errors = validator(evolved, 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=evolved.TYPE_CHECKER, ) if k not in {"if", "$ref"}: error.schema_path.appendleft(k) 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 _validate_reference(self, ref, instance): if self._ref_resolver is None: try: resolved = self._resolver.lookup(ref) except referencing.exceptions.Unresolvable as err: raise exceptions._WrappedReferencingError(err) return self.descend( instance, resolved.contents, resolver=resolved.resolver, ) else: resolve = getattr(self._ref_resolver, "resolve", None) if resolve is None: with self._ref_resolver.resolving(ref) as resolved: return self.descend(instance, resolved) else: scope, resolved = resolve(ref) self._ref_resolver.push_scope(scope) try: return list(self.descend(instance, resolved)) finally: self._ref_resolver.pop_scope() 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 evolve_fields = [ (, field.alias) for field in fields(Validator) if field.init ] if version is not None: safe = version.title().replace(" ", "").replace("-", "") Validator.__name__ = Validator.__qualname__ = f"{safe}Validator" Validator = validates(version)(Validator) # type: ignore[misc] return Validator
[docs] 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 ( 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, applicable_validators=validator._APPLICABLE_VALIDATORS, )
Draft3Validator = create( meta_schema=SPECIFICATIONS.contents( "", ), validators={ "$ref": _keywords.ref, "additionalItems": _keywords.additionalItems, "additionalProperties": _keywords.additionalProperties, "dependencies": _legacy_keywords.dependencies_draft3, "disallow": _legacy_keywords.disallow_draft3, "divisibleBy": _keywords.multipleOf, "enum": _keywords.enum, "extends": _legacy_keywords.extends_draft3, "format": _keywords.format, "items": _legacy_keywords.items_draft3_draft4, "maxItems": _keywords.maxItems, "maxLength": _keywords.maxLength, "maximum": _legacy_keywords.maximum_draft3_draft4, "minItems": _keywords.minItems, "minLength": _keywords.minLength, "minimum": _legacy_keywords.minimum_draft3_draft4, "pattern": _keywords.pattern, "patternProperties": _keywords.patternProperties, "properties": _legacy_keywords.properties_draft3, "type": _legacy_keywords.type_draft3, "uniqueItems": _keywords.uniqueItems, }, type_checker=_types.draft3_type_checker, format_checker=_format.draft3_format_checker, version="draft3", id_of=referencing.jsonschema.DRAFT3.id_of, applicable_validators=_legacy_keywords.ignore_ref_siblings, ) Draft4Validator = create( meta_schema=SPECIFICATIONS.contents( "", ), validators={ "$ref": _keywords.ref, "additionalItems": _keywords.additionalItems, "additionalProperties": _keywords.additionalProperties, "allOf": _keywords.allOf, "anyOf": _keywords.anyOf, "dependencies": _legacy_keywords.dependencies_draft4_draft6_draft7, "enum": _keywords.enum, "format": _keywords.format, "items": _legacy_keywords.items_draft3_draft4, "maxItems": _keywords.maxItems, "maxLength": _keywords.maxLength, "maxProperties": _keywords.maxProperties, "maximum": _legacy_keywords.maximum_draft3_draft4, "minItems": _keywords.minItems, "minLength": _keywords.minLength, "minProperties": _keywords.minProperties, "minimum": _legacy_keywords.minimum_draft3_draft4, "multipleOf": _keywords.multipleOf, "not": _keywords.not_, "oneOf": _keywords.oneOf, "pattern": _keywords.pattern, "patternProperties": _keywords.patternProperties, "properties":, "required": _keywords.required, "type": _keywords.type, "uniqueItems": _keywords.uniqueItems, }, type_checker=_types.draft4_type_checker, format_checker=_format.draft4_format_checker, version="draft4", id_of=referencing.jsonschema.DRAFT4.id_of, applicable_validators=_legacy_keywords.ignore_ref_siblings, ) Draft6Validator = create( meta_schema=SPECIFICATIONS.contents( "", ), validators={ "$ref": _keywords.ref, "additionalItems": _keywords.additionalItems, "additionalProperties": _keywords.additionalProperties, "allOf": _keywords.allOf, "anyOf": _keywords.anyOf, "const": _keywords.const, "contains": _legacy_keywords.contains_draft6_draft7, "dependencies": _legacy_keywords.dependencies_draft4_draft6_draft7, "enum": _keywords.enum, "exclusiveMaximum": _keywords.exclusiveMaximum, "exclusiveMinimum": _keywords.exclusiveMinimum, "format": _keywords.format, "items": _legacy_keywords.items_draft6_draft7_draft201909, "maxItems": _keywords.maxItems, "maxLength": _keywords.maxLength, "maxProperties": _keywords.maxProperties, "maximum": _keywords.maximum, "minItems": _keywords.minItems, "minLength": _keywords.minLength, "minProperties": _keywords.minProperties, "minimum": _keywords.minimum, "multipleOf": _keywords.multipleOf, "not": _keywords.not_, "oneOf": _keywords.oneOf, "pattern": _keywords.pattern, "patternProperties": _keywords.patternProperties, "properties":, "propertyNames": _keywords.propertyNames, "required": _keywords.required, "type": _keywords.type, "uniqueItems": _keywords.uniqueItems, }, type_checker=_types.draft6_type_checker, format_checker=_format.draft6_format_checker, version="draft6", id_of=referencing.jsonschema.DRAFT6.id_of, applicable_validators=_legacy_keywords.ignore_ref_siblings, ) Draft7Validator = create( meta_schema=SPECIFICATIONS.contents( "", ), validators={ "$ref": _keywords.ref, "additionalItems": _keywords.additionalItems, "additionalProperties": _keywords.additionalProperties, "allOf": _keywords.allOf, "anyOf": _keywords.anyOf, "const": _keywords.const, "contains": _legacy_keywords.contains_draft6_draft7, "dependencies": _legacy_keywords.dependencies_draft4_draft6_draft7, "enum": _keywords.enum, "exclusiveMaximum": _keywords.exclusiveMaximum, "exclusiveMinimum": _keywords.exclusiveMinimum, "format": _keywords.format, "if": _keywords.if_, "items": _legacy_keywords.items_draft6_draft7_draft201909, "maxItems": _keywords.maxItems, "maxLength": _keywords.maxLength, "maxProperties": _keywords.maxProperties, "maximum": _keywords.maximum, "minItems": _keywords.minItems, "minLength": _keywords.minLength, "minProperties": _keywords.minProperties, "minimum": _keywords.minimum, "multipleOf": _keywords.multipleOf, "not": _keywords.not_, "oneOf": _keywords.oneOf, "pattern": _keywords.pattern, "patternProperties": _keywords.patternProperties, "properties":, "propertyNames": _keywords.propertyNames, "required": _keywords.required, "type": _keywords.type, "uniqueItems": _keywords.uniqueItems, }, type_checker=_types.draft7_type_checker, format_checker=_format.draft7_format_checker, version="draft7", id_of=referencing.jsonschema.DRAFT7.id_of, applicable_validators=_legacy_keywords.ignore_ref_siblings, ) Draft201909Validator = create( meta_schema=SPECIFICATIONS.contents( "", ), validators={ "$recursiveRef": _legacy_keywords.recursiveRef, "$ref": _keywords.ref, "additionalItems": _keywords.additionalItems, "additionalProperties": _keywords.additionalProperties, "allOf": _keywords.allOf, "anyOf": _keywords.anyOf, "const": _keywords.const, "contains": _keywords.contains, "dependentRequired": _keywords.dependentRequired, "dependentSchemas": _keywords.dependentSchemas, "enum": _keywords.enum, "exclusiveMaximum": _keywords.exclusiveMaximum, "exclusiveMinimum": _keywords.exclusiveMinimum, "format": _keywords.format, "if": _keywords.if_, "items": _legacy_keywords.items_draft6_draft7_draft201909, "maxItems": _keywords.maxItems, "maxLength": _keywords.maxLength, "maxProperties": _keywords.maxProperties, "maximum": _keywords.maximum, "minItems": _keywords.minItems, "minLength": _keywords.minLength, "minProperties": _keywords.minProperties, "minimum": _keywords.minimum, "multipleOf": _keywords.multipleOf, "not": _keywords.not_, "oneOf": _keywords.oneOf, "pattern": _keywords.pattern, "patternProperties": _keywords.patternProperties, "properties":, "propertyNames": _keywords.propertyNames, "required": _keywords.required, "type": _keywords.type, "unevaluatedItems": _legacy_keywords.unevaluatedItems_draft2019, "unevaluatedProperties": _keywords.unevaluatedProperties, "uniqueItems": _keywords.uniqueItems, }, type_checker=_types.draft201909_type_checker, format_checker=_format.draft201909_format_checker, version="draft2019-09", ) Draft202012Validator = create( meta_schema=SPECIFICATIONS.contents( "", ), validators={ "$dynamicRef": _keywords.dynamicRef, "$ref": _keywords.ref, "additionalItems": _keywords.additionalItems, "additionalProperties": _keywords.additionalProperties, "allOf": _keywords.allOf, "anyOf": _keywords.anyOf, "const": _keywords.const, "contains": _keywords.contains, "dependentRequired": _keywords.dependentRequired, "dependentSchemas": _keywords.dependentSchemas, "enum": _keywords.enum, "exclusiveMaximum": _keywords.exclusiveMaximum, "exclusiveMinimum": _keywords.exclusiveMinimum, "format": _keywords.format, "if": _keywords.if_, "items": _keywords.items, "maxItems": _keywords.maxItems, "maxLength": _keywords.maxLength, "maxProperties": _keywords.maxProperties, "maximum": _keywords.maximum, "minItems": _keywords.minItems, "minLength": _keywords.minLength, "minProperties": _keywords.minProperties, "minimum": _keywords.minimum, "multipleOf": _keywords.multipleOf, "not": _keywords.not_, "oneOf": _keywords.oneOf, "pattern": _keywords.pattern, "patternProperties": _keywords.patternProperties, "prefixItems": _keywords.prefixItems, "properties":, "propertyNames": _keywords.propertyNames, "required": _keywords.required, "type": _keywords.type, "unevaluatedItems": _keywords.unevaluatedItems, "unevaluatedProperties": _keywords.unevaluatedProperties, "uniqueItems": _keywords.uniqueItems, }, type_checker=_types.draft202012_type_checker, format_checker=_format.draft202012_format_checker, version="draft2020-12", ) _LATEST_VERSION = Draft202012Validator
[docs] 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 .. deprecated:: v4.18.0 ``RefResolver`` has been deprecated in favor of `referencing`. """ _DEPRECATION_MESSAGE = ( "jsonschema.RefResolver is deprecated as of v4.18.0, in favor of the " " library, which " "provides more compliant referencing behavior as well as more " "flexible APIs for customization. A future release will remove " "RefResolver. Please file a feature request (on referencing) if you " "are missing an API for the kind of customization you need." ) def __init__( self, base_uri, referrer, store=HashTrieMap(), 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] = _utils.URIDict( (uri, each.contents) for uri, each in SPECIFICATIONS.items() ) (id, each.META_SCHEMA) for id, each in _META_SCHEMAS.items() ) (schema["$id"], schema) for schema in store.values() if isinstance(schema, Mapping) and "$id" in schema )[base_uri] = referrer self._urljoin_cache = urljoin_cache self._remote_cache = remote_cache
[docs] @classmethod def from_schema( # noqa: D417 cls, schema, id_of=referencing.jsonschema.DRAFT202012.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) or "", referrer=schema, *args, **kwargs) # noqa: B026, E501
[docs] 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), )
[docs] 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
[docs] @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()
[docs] @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: id = subschema["$id"] if not isinstance(id, str): continue target_uri = self._urljoin_cache(self.resolution_scope, id) if target_uri.rstrip("/") == uri.rstrip("/"): if fragment: subschema = self.resolve_fragment(subschema, fragment)[url] = subschema return url, subschema return None
[docs] 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)
[docs] def resolve_from_url(self, url): """ Resolve the given URL. """ url, fragment = urldefrag(url) if not url: url = self.base_uri try: document =[url] except KeyError: try: document = self.resolve_remote(url) except Exception as exc: raise exceptions._RefResolutionError(exc) return self.resolve_fragment(document, fragment)
[docs] 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): try: # noqa: SIM105 part = int(part) except ValueError: pass try: document = document[part] except (TypeError, LookupError): raise exceptions._RefResolutionError( f"Unresolvable JSON pointer: {fragment!r}", ) return document
[docs] 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: """ 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("utf-8")) if self.cache_remote:[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())
[docs] def validate(instance, schema, cls=None, *args, **kwargs): # noqa: D417 """ 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. ``Draft202012Validator.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
[docs] 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 ( 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. Examples: The :kw:`$schema` JSON Schema keyword will control which validator class is returned: >>> schema = { ... "$schema": "", ... "type": "integer", ... } >>> jsonschema.validators.validator_for(schema) <class 'jsonschema.validators.Draft202012Validator'> Here, a draft 7 schema instead will return the draft 7 validator: >>> schema = { ... "$schema": "", ... "type": "integer", ... } >>> jsonschema.validators.validator_for(schema) <class 'jsonschema.validators.Draft7Validator'> Schemas with no ``$schema`` keyword will fallback to the default argument: >>> schema = {"type": "integer"} >>> jsonschema.validators.validator_for( ... schema, default=Draft7Validator, ... ) <class 'jsonschema.validators.Draft7Validator'> or if none is provided, to the latest version supported. Always including the keyword when authoring schemas is highly recommended. """ 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 and 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)