schema.py

"""schema is a library for validating Python data structures, such as those
obtained from config-files, forms, external services or command-line
parsing, converted from JSON/YAML (or something else) to Python data-types."""

import inspect
import re
from typing import Any
from typing import Callable
from typing import cast
from typing import Dict
from typing import Generic
from typing import Iterable
from typing import List
from typing import NoReturn
from typing import Sequence
from typing import Set
from typing import Sized
from typing import Tuple
from typing import Type
from typing import TYPE_CHECKING
from typing import TypeVar
from typing import Union


# Use TYPE_CHECKING to determine the correct type hint but avoid runtime import errors
if TYPE_CHECKING:
    # Only for type checking purposes, we import the standard ExitStack
    from contextlib import ExitStack
else:
    try:
        from contextlib import ExitStack  # Python 3.3 and later
    except ImportError:
        from contextlib2 import ExitStack  # Python 2.x/3.0-3.2 fallback


__version__ = "0.7.7"
__all__ = [
    "Schema",
    "And",
    "Or",
    "Regex",
    "Optional",
    "Use",
    "Forbidden",
    "Const",
    "Literal",
    "SchemaError",
    "SchemaWrongKeyError",
    "SchemaMissingKeyError",
    "SchemaForbiddenKeyError",
    "SchemaUnexpectedTypeError",
    "SchemaOnlyOneAllowedError",
]


class SchemaError(Exception):
    """Error during Schema validation."""

    def __init__(
        self,
        autos: Union[Sequence[Union[str, None]], None],
        errors: Union[List, str, None] = None,
    ):
        self.autos = autos if isinstance(autos, List) else [autos]
        self.errors = errors if isinstance(errors, List) else [errors]
        Exception.__init__(self, self.code)

    @property
    def code(self) -> str:
        """Remove duplicates in autos and errors list and combine them into a single message."""

        def uniq(seq: Iterable[Union[str, None]]) -> List[str]:
            """Utility function to remove duplicates while preserving the order."""
            seen: Set[str] = set()
            unique_list: List[str] = []
            for x in seq:
                if x is not None and x not in seen:
                    seen.add(x)
                    unique_list.append(x)
            return unique_list

        data_set = uniq(self.autos)
        error_list = uniq(self.errors)

        return "\n".join(error_list if error_list else data_set)


class SchemaWrongKeyError(SchemaError):
    """Error Should be raised when an unexpected key is detected within the
    data set being."""

    pass


class SchemaMissingKeyError(SchemaError):
    """Error should be raised when a mandatory key is not found within the
    data set being validated"""

    pass


class SchemaOnlyOneAllowedError(SchemaError):
    """Error should be raised when an only_one Or key has multiple matching candidates"""

    pass


class SchemaForbiddenKeyError(SchemaError):
    """Error should be raised when a forbidden key is found within the
    data set being validated, and its value matches the value that was specified"""

    pass


class SchemaUnexpectedTypeError(SchemaError):
    """Error should be raised when a type mismatch is detected within the
    data set being validated."""

    pass


# Type variable to represent a Schema-like type
TSchema = TypeVar("TSchema", bound="Schema")


class And(Generic[TSchema]):
    """
    Utility function to combine validation directives in AND Boolean fashion.
    """

    def __init__(
        self,
        *args: Union[TSchema, Callable[..., Any]],
        error: Union[str, None] = None,
        ignore_extra_keys: bool = False,
        schema: Union[Type[TSchema], None] = None,
    ) -> None:
        self._args: Tuple[Union[TSchema, Callable[..., Any]], ...] = args
        self._error: Union[str, None] = error
        self._ignore_extra_keys: bool = ignore_extra_keys
        self._schema_class: Type[TSchema] = schema if schema is not None else Schema

    def __repr__(self) -> str:
        return f"{self.__class__.__name__}({', '.join(repr(a) for a in self._args)})"

    @property
    def args(self) -> Tuple[Union[TSchema, Callable[..., Any]], ...]:
        """The provided parameters"""
        return self._args

    def validate(self, data: Any, **kwargs: Any) -> Any:
        """
        Validate data using defined sub schema/expressions ensuring all
        values are valid.
        :param data: Data to be validated with sub defined schemas.
        :return: Returns validated data.
        """
        # Annotate sub_schema with the type returned by _build_schema
        for sub_schema in self._build_schemas():  # type: TSchema
            data = sub_schema.validate(data, **kwargs)
        return data

    def _build_schemas(self) -> List[TSchema]:
        return [self._build_schema(s) for s in self._args]

    def _build_schema(self, arg: Any) -> TSchema:
        # Assume self._schema_class(arg, ...) returns an instance of TSchema
        return self._schema_class(
            arg, error=self._error, ignore_extra_keys=self._ignore_extra_keys
        )


class Or(And[TSchema]):
    """Utility function to combine validation directives in a OR Boolean
    fashion.

    If one wants to make an xor, one can provide only_one=True optional argument
    to the constructor of this object. When a validation was performed for an
    xor-ish Or instance and one wants to use it another time, one needs to call
    reset() to put the match_count back to 0."""

    def __init__(
        self,
        *args: Union[TSchema, Callable[..., Any]],
        only_one: bool = False,
        **kwargs: Any,
    ) -> None:
        self.only_one: bool = only_one
        self.match_count: int = 0
        super().__init__(*args, **kwargs)

    def reset(self) -> None:
        failed: bool = self.match_count > 1 and self.only_one
        self.match_count = 0
        if failed:
            raise SchemaOnlyOneAllowedError(
                ["There are multiple keys present from the %r condition" % self]
            )

    def validate(self, data: Any, **kwargs: Any) -> Any:
        """
        Validate data using sub defined schema/expressions ensuring at least
        one value is valid.
        :param data: data to be validated by provided schema.
        :return: return validated data if not validation
        """
        autos: List[str] = []
        errors: List[Union[str, None]] = []
        for sub_schema in self._build_schemas():
            try:
                validation: Any = sub_schema.validate(data, **kwargs)
                self.match_count += 1
                if self.match_count > 1 and self.only_one:
                    break
                return validation
            except SchemaError as _x:
                autos += _x.autos
                errors += _x.errors
        raise SchemaError(
            ["%r did not validate %r" % (self, data)] + autos,
            [self._error.format(data) if self._error else None] + errors,
        )


class Regex:
    """
    Enables schema.py to validate string using regular expressions.
    """

    # Map all flags bits to a more readable description
    NAMES = [
        "re.ASCII",
        "re.DEBUG",
        "re.VERBOSE",
        "re.UNICODE",
        "re.DOTALL",
        "re.MULTILINE",
        "re.LOCALE",
        "re.IGNORECASE",
        "re.TEMPLATE",
    ]

    def __init__(
        self, pattern_str: str, flags: int = 0, error: Union[str, None] = None
    ) -> None:
        self._pattern_str: str = pattern_str
        flags_list = [
            Regex.NAMES[i] for i, f in enumerate(f"{flags:09b}") if f != "0"
        ]  # Name for each bit

        self._flags_names: str = ", flags=" + "|".join(flags_list) if flags_list else ""
        self._pattern: re.Pattern = re.compile(pattern_str, flags=flags)
        self._error: Union[str, None] = error

    def __repr__(self) -> str:
        return f"{self.__class__.__name__}({self._pattern_str!r}{self._flags_names})"

    @property
    def pattern_str(self) -> str:
        """The pattern string for the represented regular expression"""
        return self._pattern_str

    def validate(self, data: str, **kwargs: Any) -> str:
        """
        Validates data using the defined regex.
        :param data: Data to be validated.
        :return: Returns validated data.
        """
        e = self._error

        try:
            if self._pattern.search(data):
                return data
            else:
                error_message = (
                    e.format(data)
                    if e
                    else f"{data!r} does not match {self._pattern_str!r}"
                )
                raise SchemaError(error_message)
        except TypeError:
            error_message = (
                e.format(data) if e else f"{data!r} is not string nor buffer"
            )
            raise SchemaError(error_message)


class Use:
    """
    For more general use cases, you can use the Use class to transform
    the data while it is being validated.
    """

    def __init__(
        self, callable_: Callable[[Any], Any], error: Union[str, None] = None
    ) -> None:
        if not callable(callable_):
            raise TypeError(f"Expected a callable, not {callable_!r}")
        self._callable: Callable[[Any], Any] = callable_
        self._error: Union[str, None] = error

    def __repr__(self) -> str:
        return f"{self.__class__.__name__}({self._callable!r})"

    def validate(self, data: Any, **kwargs: Any) -> Any:
        try:
            return self._callable(data)
        except SchemaError as x:
            raise SchemaError(
                [None] + x.autos,
                [self._error.format(data) if self._error else None] + x.errors,
            )
        except BaseException as x:
            f = _callable_str(self._callable)
            raise SchemaError(
                "%s(%r) raised %r" % (f, data, x),
                self._error.format(data) if self._error else None,
            )


COMPARABLE, CALLABLE, VALIDATOR, TYPE, DICT, ITERABLE = range(6)


def _priority(s: Any) -> int:
    """Return priority for a given object."""
    if type(s) in (list, tuple, set, frozenset):
        return ITERABLE
    if isinstance(s, dict):
        return DICT
    if issubclass(type(s), type):
        return TYPE
    if isinstance(s, Literal):
        return COMPARABLE
    if hasattr(s, "validate"):
        return VALIDATOR
    if callable(s):
        return CALLABLE
    else:
        return COMPARABLE


def _invoke_with_optional_kwargs(f: Callable[..., Any], **kwargs: Any) -> Any:
    s = inspect.signature(f)
    if len(s.parameters) == 0:
        return f()
    return f(**kwargs)


class Schema(object):
    """
    Entry point of the library, use this class to instantiate validation
    schema for the data that will be validated.
    """

    def __init__(
        self,
        schema: Any,
        error: Union[str, None] = None,
        ignore_extra_keys: bool = False,
        name: Union[str, None] = None,
        description: Union[str, None] = None,
        as_reference: bool = False,
    ) -> None:
        self._schema: Any = schema
        self._error: Union[str, None] = error
        self._ignore_extra_keys: bool = ignore_extra_keys
        self._name: Union[str, None] = name
        self._description: Union[str, None] = description
        self.as_reference: bool = as_reference

        if as_reference and name is None:
            raise ValueError("Schema used as reference should have a name")

    def __repr__(self):
        return "%s(%r)" % (self.__class__.__name__, self._schema)

    @property
    def schema(self) -> Any:
        return self._schema

    @property
    def description(self) -> Union[str, None]:
        return self._description

    @property
    def name(self) -> Union[str, None]:
        return self._name

    @property
    def ignore_extra_keys(self) -> bool:
        return self._ignore_extra_keys

    @staticmethod
    def _dict_key_priority(s) -> float:
        """Return priority for a given key object."""
        if isinstance(s, Hook):
            return _priority(s._schema) - 0.5
        if isinstance(s, Optional):
            return _priority(s._schema) + 0.5
        return _priority(s)

    @staticmethod
    def _is_optional_type(s: Any) -> bool:
        """Return True if the given key is optional (does not have to be found)"""
        return any(isinstance(s, optional_type) for optional_type in [Optional, Hook])

    def is_valid(self, data: Any, **kwargs: Dict[str, Any]) -> bool:
        """Return whether the given data has passed all the validations
        that were specified in the given schema.
        """
        try:
            self.validate(data, **kwargs)
        except SchemaError:
            return False
        else:
            return True

    def _prepend_schema_name(self, message: str) -> str:
        """
        If a custom schema name has been defined, prepends it to the error
        message that gets raised when a schema error occurs.
        """
        if self._name:
            message = "{0!r} {1!s}".format(self._name, message)
        return message

    def validate(self, data: Any, **kwargs: Dict[str, Any]) -> Any:
        Schema = self.__class__
        s: Any = self._schema
        e: Union[str, None] = self._error
        i: bool = self._ignore_extra_keys

        if isinstance(s, Literal):
            s = s.schema

        flavor = _priority(s)
        if flavor == ITERABLE:
            data = Schema(type(s), error=e).validate(data, **kwargs)
            o: Or = Or(*s, error=e, schema=Schema, ignore_extra_keys=i)
            return type(data)(o.validate(d, **kwargs) for d in data)
        if flavor == DICT:
            exitstack = ExitStack()
            data = Schema(dict, error=e).validate(data, **kwargs)
            new: Dict = type(data)()  # new - is a dict of the validated values
            coverage: Set = set()  # matched schema keys
            # for each key and value find a schema entry matching them, if any
            sorted_skeys = sorted(s, key=self._dict_key_priority)
            for skey in sorted_skeys:
                if hasattr(skey, "reset"):
                    exitstack.callback(skey.reset)

            with exitstack:
                # Evaluate dictionaries last
                data_items = sorted(
                    data.items(), key=lambda value: isinstance(value[1], dict)
                )
                for key, value in data_items:
                    for skey in sorted_skeys:
                        svalue = s[skey]
                        try:
                            nkey = Schema(skey, error=e).validate(key, **kwargs)
                        except SchemaError:
                            pass
                        else:
                            if isinstance(skey, Hook):
                                # As the content of the value makes little sense for
                                # keys with a hook, we reverse its meaning:
                                # we will only call the handler if the value does match
                                # In the case of the forbidden key hook,
                                # we will raise the SchemaErrorForbiddenKey exception
                                # on match, allowing for excluding a key only if its
                                # value has a certain type, and allowing Forbidden to
                                # work well in combination with Optional.
                                try:
                                    nvalue = Schema(svalue, error=e).validate(
                                        value, **kwargs
                                    )
                                except SchemaError:
                                    continue
                                skey.handler(nkey, data, e)
                            else:
                                try:
                                    nvalue = Schema(
                                        svalue, error=e, ignore_extra_keys=i
                                    ).validate(value, **kwargs)
                                except SchemaError as x:
                                    k = "Key '%s' error:" % nkey
                                    message = self._prepend_schema_name(k)
                                    raise SchemaError(
                                        [message] + x.autos,
                                        [e.format(data) if e else None] + x.errors,
                                    )
                                else:
                                    new[nkey] = nvalue
                                    coverage.add(skey)
                                    break
            required = set(k for k in s if not self._is_optional_type(k))
            if not required.issubset(coverage):
                missing_keys = required - coverage
                s_missing_keys = ", ".join(
                    repr(k) for k in sorted(missing_keys, key=repr)
                )
                message = "Missing key%s: %s" % (
                    _plural_s(missing_keys),
                    s_missing_keys,
                )
                message = self._prepend_schema_name(message)
                raise SchemaMissingKeyError(message, e.format(data) if e else None)
            if not self._ignore_extra_keys and (len(new) != len(data)):
                wrong_keys = set(data.keys()) - set(new.keys())
                s_wrong_keys = ", ".join(repr(k) for k in sorted(wrong_keys, key=repr))
                message = "Wrong key%s %s in %r" % (
                    _plural_s(wrong_keys),
                    s_wrong_keys,
                    data,
                )
                message = self._prepend_schema_name(message)
                raise SchemaWrongKeyError(message, e.format(data) if e else None)

            # Apply default-having optionals that haven't been used:
            defaults = (
                set(k for k in s if isinstance(k, Optional) and hasattr(k, "default"))
                - coverage
            )
            for default in defaults:
                new[default.key] = (
                    _invoke_with_optional_kwargs(default.default, **kwargs)
                    if callable(default.default)
                    else default.default
                )

            return new
        if flavor == TYPE:
            if isinstance(data, s) and not (isinstance(data, bool) and s == int):
                return data
            else:
                message = "%r should be instance of %r" % (data, s.__name__)
                message = self._prepend_schema_name(message)
                raise SchemaUnexpectedTypeError(message, e.format(data) if e else None)
        if flavor == VALIDATOR:
            try:
                return s.validate(data, **kwargs)
            except SchemaError as x:
                raise SchemaError(
                    [None] + x.autos, [e.format(data) if e else None] + x.errors
                )
            except BaseException as x:
                message = "%r.validate(%r) raised %r" % (s, data, x)
                message = self._prepend_schema_name(message)
                raise SchemaError(message, e.format(data) if e else None)
        if flavor == CALLABLE:
            f = _callable_str(s)
            try:
                if s(data):
                    return data
            except SchemaError as x:
                raise SchemaError(
                    [None] + x.autos, [e.format(data) if e else None] + x.errors
                )
            except BaseException as x:
                message = "%s(%r) raised %r" % (f, data, x)
                message = self._prepend_schema_name(message)
                raise SchemaError(message, e.format(data) if e else None)
            message = "%s(%r) should evaluate to True" % (f, data)
            message = self._prepend_schema_name(message)
            raise SchemaError(message, e.format(data) if e else None)
        if s == data:
            return data
        else:
            message = "%r does not match %r" % (s, data)
            message = self._prepend_schema_name(message)
            raise SchemaError(message, e.format(data) if e else None)

    def json_schema(
        self, schema_id: str, use_refs: bool = False, **kwargs: Any
    ) -> Dict[str, Any]:
        """Generate a draft-07 JSON schema dict representing the Schema.
        This method must be called with a schema_id.

        :param schema_id: The value of the $id on the main schema
        :param use_refs: Enable reusing object references in the resulting JSON schema.
                         Schemas with references are harder to read by humans, but are a lot smaller when there
                         is a lot of reuse
        """

        seen: Dict[int, Dict[str, Any]] = {}
        definitions_by_name: Dict[str, Dict[str, Any]] = {}

        def _json_schema(
            schema: "Schema",
            is_main_schema: bool = True,
            description: Union[str, None] = None,
            allow_reference: bool = True,
        ) -> Dict[str, Any]:
            def _create_or_use_ref(return_dict: Dict[str, Any]) -> Dict[str, Any]:
                """If not already seen, return the provided part of the schema unchanged.
                If already seen, give an id to the already seen dict and return a reference to the previous part
                of the schema instead.
                """
                if not use_refs or is_main_schema:
                    return return_schema

                hashed = hash(repr(sorted(return_dict.items())))
                if hashed not in seen:
                    seen[hashed] = return_dict
                    return return_dict
                else:
                    id_str = "#" + str(hashed)
                    seen[hashed]["$id"] = id_str
                    return {"$ref": id_str}

            def _get_type_name(python_type: Type) -> str:
                """Return the JSON schema name for a Python type"""
                if python_type == str:
                    return "string"
                elif python_type == int:
                    return "integer"
                elif python_type == float:
                    return "number"
                elif python_type == bool:
                    return "boolean"
                elif python_type == list:
                    return "array"
                elif python_type == dict:
                    return "object"
                return "string"

            def _to_json_type(value: Any) -> Any:
                """Attempt to convert a constant value (for "const" and "default") to a JSON serializable value"""
                if value is None or type(value) in (str, int, float, bool, list, dict):
                    return value

                if type(value) in (tuple, set, frozenset):
                    return list(value)

                if isinstance(value, Literal):
                    return value.schema

                return str(value)

            def _to_schema(s: Any, ignore_extra_keys: bool) -> Schema:
                if not isinstance(s, Schema):
                    return Schema(s, ignore_extra_keys=ignore_extra_keys)

                return s

            s: Any = schema.schema
            i: bool = schema.ignore_extra_keys
            flavor = _priority(s)

            return_schema: Dict[str, Any] = {}

            return_description: Union[str, None] = description or schema.description
            if return_description:
                return_schema["description"] = return_description

            # Check if we have to create a common definition and use as reference
            if allow_reference and schema.as_reference:
                # Generate sub schema if not already done
                if schema.name not in definitions_by_name:
                    definitions_by_name[
                        cast(str, schema.name)
                    ] = {}  # Avoid infinite loop
                    definitions_by_name[cast(str, schema.name)] = _json_schema(
                        schema, is_main_schema=False, allow_reference=False
                    )

                return_schema["$ref"] = "#/definitions/" + cast(str, schema.name)
            else:
                if flavor == TYPE:
                    # Handle type
                    return_schema["type"] = _get_type_name(s)
                elif flavor == ITERABLE:
                    # Handle arrays or dict schema

                    return_schema["type"] = "array"
                    if len(s) == 1:
                        return_schema["items"] = _json_schema(
                            _to_schema(s[0], i), is_main_schema=False
                        )
                    elif len(s) > 1:
                        return_schema["items"] = _json_schema(
                            Schema(Or(*s)), is_main_schema=False
                        )
                elif isinstance(s, Or):
                    # Handle Or values

                    # Check if we can use an enum
                    if all(
                        priority == COMPARABLE
                        for priority in [_priority(value) for value in s.args]
                    ):
                        or_values = [
                            str(s) if isinstance(s, Literal) else s for s in s.args
                        ]
                        # All values are simple, can use enum or const
                        if len(or_values) == 1:
                            return_schema["const"] = _to_json_type(or_values[0])
                            return return_schema
                        return_schema["enum"] = or_values
                    else:
                        # No enum, let's go with recursive calls
                        any_of_values = []
                        for or_key in s.args:
                            new_value = _json_schema(
                                _to_schema(or_key, i), is_main_schema=False
                            )
                            if new_value != {} and new_value not in any_of_values:
                                any_of_values.append(new_value)
                        if len(any_of_values) == 1:
                            # Only one representable condition remains, do not put under anyOf
                            return_schema.update(any_of_values[0])
                        else:
                            return_schema["anyOf"] = any_of_values
                elif isinstance(s, And):
                    # Handle And values
                    all_of_values = []
                    for and_key in s.args:
                        new_value = _json_schema(
                            _to_schema(and_key, i), is_main_schema=False
                        )
                        if new_value != {} and new_value not in all_of_values:
                            all_of_values.append(new_value)
                    if len(all_of_values) == 1:
                        # Only one representable condition remains, do not put under allOf
                        return_schema.update(all_of_values[0])
                    else:
                        return_schema["allOf"] = all_of_values
                elif flavor == COMPARABLE:
                    return_schema["const"] = _to_json_type(s)
                elif flavor == VALIDATOR and type(s) == Regex:
                    return_schema["type"] = "string"
                    return_schema["pattern"] = s.pattern_str
                else:
                    if flavor != DICT:
                        # If not handled, do not check
                        return return_schema

                    # Schema is a dict

                    required_keys = []
                    expanded_schema = {}
                    additional_properties = i
                    for key in s:
                        if isinstance(key, Hook):
                            continue

                        def _key_allows_additional_properties(key: Any) -> bool:
                            """Check if a key is broad enough to allow additional properties"""
                            if isinstance(key, Optional):
                                return _key_allows_additional_properties(key.schema)

                            return key == str or key == object

                        def _get_key_description(key: Any) -> Union[str, None]:
                            """Get the description associated to a key (as specified in a Literal object). Return None if not a Literal"""
                            if isinstance(key, Optional):
                                return _get_key_description(key.schema)

                            if isinstance(key, Literal):
                                return key.description

                            return None

                        def _get_key_name(key: Any) -> Any:
                            """Get the name of a key (as specified in a Literal object). Return the key unchanged if not a Literal"""
                            if isinstance(key, Optional):
                                return _get_key_name(key.schema)

                            if isinstance(key, Literal):
                                return key.schema

                            return key

                        additional_properties = (
                            additional_properties
                            or _key_allows_additional_properties(key)
                        )
                        sub_schema = _to_schema(s[key], ignore_extra_keys=i)
                        key_name = _get_key_name(key)

                        if isinstance(key_name, str):
                            if not isinstance(key, Optional):
                                required_keys.append(key_name)
                            expanded_schema[key_name] = _json_schema(
                                sub_schema,
                                is_main_schema=False,
                                description=_get_key_description(key),
                            )
                            if isinstance(key, Optional) and hasattr(key, "default"):
                                expanded_schema[key_name]["default"] = _to_json_type(
                                    _invoke_with_optional_kwargs(key.default, **kwargs)
                                    if callable(key.default)
                                    else key.default
                                )
                        elif isinstance(key_name, Or):
                            # JSON schema does not support having a key named one name or another, so we just add both options
                            # This is less strict because we cannot enforce that one or the other is required

                            for or_key in key_name.args:
                                expanded_schema[_get_key_name(or_key)] = _json_schema(
                                    sub_schema,
                                    is_main_schema=False,
                                    description=_get_key_description(or_key),
                                )

                    return_schema.update(
                        {
                            "type": "object",
                            "properties": expanded_schema,
                            "required": required_keys,
                            "additionalProperties": additional_properties,
                        }
                    )

            if is_main_schema:
                return_schema.update(
                    {
                        "$id": schema_id,
                        "$schema": "http://json-schema.org/draft-07/schema#",
                    }
                )
                if self._name:
                    return_schema["title"] = self._name

                if definitions_by_name:
                    return_schema["definitions"] = {}
                    for definition_name, definition in definitions_by_name.items():
                        return_schema["definitions"][definition_name] = definition

            return _create_or_use_ref(return_schema)

        return _json_schema(self, True)


class Optional(Schema):
    """Marker for an optional part of the validation Schema."""

    _MARKER = object()

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        default: Any = kwargs.pop("default", self._MARKER)
        super(Optional, self).__init__(*args, **kwargs)
        if default is not self._MARKER:
            if _priority(self._schema) != COMPARABLE:
                raise TypeError(
                    "Optional keys with defaults must have simple, "
                    "predictable values, like literal strings or ints. "
                    f'"{self._schema!r}" is too complex.'
                )
            self.default = default
            self.key = str(self._schema)

    def __hash__(self) -> int:
        return hash(self._schema)

    def __eq__(self, other: Any) -> bool:
        return (
            self.__class__ is other.__class__
            and getattr(self, "default", self._MARKER)
            == getattr(other, "default", self._MARKER)
            and self._schema == other._schema
        )

    def reset(self) -> None:
        if hasattr(self._schema, "reset"):
            self._schema.reset()


class Hook(Schema):
    def __init__(self, *args: Any, **kwargs: Any) -> None:
        self.handler: Callable[..., Any] = kwargs.pop("handler", lambda *args: None)
        super(Hook, self).__init__(*args, **kwargs)
        self.key = self._schema


class Forbidden(Hook):
    def __init__(self, *args: Any, **kwargs: Any) -> None:
        kwargs["handler"] = self._default_function
        super(Forbidden, self).__init__(*args, **kwargs)

    @staticmethod
    def _default_function(nkey: Any, data: Any, error: Any) -> NoReturn:
        raise SchemaForbiddenKeyError(
            f"Forbidden key encountered: {nkey!r} in {data!r}", error
        )


class Literal:
    def __init__(self, value: Any, description: Union[str, None] = None) -> None:
        self._schema: Any = value
        self._description: Union[str, None] = description

    def __str__(self) -> str:
        return str(self._schema)

    def __repr__(self) -> str:
        return f'Literal("{self._schema}", description="{self._description or ""}")'

    @property
    def description(self) -> Union[str, None]:
        return self._description

    @property
    def schema(self) -> Any:
        return self._schema


class Const(Schema):
    def validate(self, data: Any, **kwargs: Any) -> Any:
        super(Const, self).validate(data, **kwargs)
        return data


def _callable_str(callable_: Callable[..., Any]) -> str:
    if hasattr(callable_, "__name__"):
        return callable_.__name__
    return str(callable_)


def _plural_s(sized: Sized) -> str:
    return "s" if len(sized) > 1 else ""