path: root/libs/jsonschema/validators.py

"""
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)