# Source code for xsdata.formats.dataclass.parsers.json

    import json
    import warnings
    from dataclasses import dataclass, field
    from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Type, Union

    from xsdata.exceptions import ConverterWarning, ParserError
    from xsdata.formats.bindings import AbstractParser, T
    from xsdata.formats.converter import converter
    from xsdata.formats.dataclass.context import XmlContext
    from xsdata.formats.dataclass.models.elements import XmlMeta, XmlVar
    from xsdata.formats.dataclass.parsers.config import ParserConfig
    from xsdata.formats.dataclass.parsers.utils import ParserUtils
    from xsdata.formats.dataclass.typing import get_args, get_origin
    from xsdata.utils import collections
    from xsdata.utils.constants import EMPTY_MAP

    @dataclass
    class JsonParser(AbstractParser):
        """
        Json parser for dataclasses.

        :param config: Parser configuration
        :param context: Model context provider
        :param load_factory: Replace the default json.load call with another
            implementation
        """

        config: ParserConfig = field(default_factory=ParserConfig)
        context: XmlContext = field(default_factory=XmlContext)
        load_factory: Callable = field(default=json.load)

        def parse(self, source: Any, clazz: Optional[Type[T]] = None) -> T:
            """Parse the input stream or filename and return the resulting object
            tree."""

            data = self.load_json(source)
            tp = self.verify_type(clazz, data)

            with warnings.catch_warnings():
                if self.config.fail_on_converter_warnings:
                    warnings.filterwarnings("error", category=ConverterWarning)

                try:
                    if not isinstance(data, list):
                        return self.bind_dataclass(data, tp)

                    return [self.bind_dataclass(obj, tp) for obj in data]  # type: ignore
                except ConverterWarning as e:
                    raise ParserError(e)

        def load_json(self, source: Any) -> Union[Dict, List]:
            if not hasattr(source, "read"):
                with open(source, "rb") as fp:
                    return self.load_factory(fp)

            return self.load_factory(source)

        def verify_type(self, clazz: Optional[Type[T]], data: Union[Dict, List]) -> Type[T]:
            if clazz is None:
                return self.detect_type(data)

            try:
                origin = get_origin(clazz)
                list_type = False
                if origin is list:
                    list_type = True
                    args = get_args(clazz)

                    if len(args) != 1 or not self.context.class_type.is_model(args[0]):
                        raise TypeError()

                    clazz = args[0]
                elif origin is not None:
                    raise TypeError()
            except TypeError:
                raise ParserError(f"Invalid clazz argument: {clazz}")

            if list_type != isinstance(data, list):
                if list_type:
                    raise ParserError("Document is object, expected array")
                raise ParserError("Document is array, expected object")

            return clazz  # type: ignore

        def detect_type(self, data: Union[Dict, List]) -> Type[T]:
            if not data:
                raise ParserError("Document is empty, can not detect type")

            keys = data[0].keys() if isinstance(data, list) else data.keys()
            clazz: Optional[Type[T]] = self.context.find_type_by_fields(set(keys))

            if clazz:
                return clazz

            raise ParserError(f"Unable to locate model with properties({list(keys)})")

        def bind_dataclass(self, data: Dict, clazz: Type[T]) -> T:
            """Recursively build the given model from the input dict data."""
            if set(data.keys()) == self.context.class_type.derived_keys:
                return self.bind_derived_dataclass(data, clazz)

            meta = self.context.build(clazz)
            xml_vars = meta.get_all_vars()

            params = {}
            for key, value in data.items():
                is_array = collections.is_array(value)
                var = self.find_var(xml_vars, key, is_array)

                if var is None and self.config.fail_on_unknown_properties:
                    raise ParserError(f"Unknown property {clazz.__qualname__}.{key}")

                if var and var.init:
                    params[var.name] = self.bind_value(meta, var, value)

            try:
                return self.config.class_factory(clazz, params)
            except TypeError as e:
                raise ParserError(e)

        def bind_derived_dataclass(self, data: Dict, clazz: Type[T]) -> Any:
            qname = data["qname"]
            xsi_type = data["type"]
            params = data["value"]

            generic = self.context.class_type.derived_element

            if clazz is generic:
                real_clazz: Optional[Type[T]] = None
                if xsi_type:
                    real_clazz = self.context.find_type(xsi_type)

                if real_clazz is None:
                    raise ParserError(
                        f"Unable to locate derived model "
                        f"with properties({list(params.keys())})"
                    )

                value = self.bind_dataclass(params, real_clazz)
            else:
                value = self.bind_dataclass(params, clazz)

            return generic(qname=qname, type=xsi_type, value=value)

        def bind_best_dataclass(self, data: Dict, classes: Iterable[Type[T]]) -> T:
            """Attempt to bind the given data to one possible models, if more than
            one is successful return the object with the highest score."""
            obj = None
            keys = set(data.keys())
            max_score = -1.0
            for clazz in classes:
                if not self.context.class_type.is_model(clazz):
                    continue

                if self.context.local_names_match(keys, clazz):
                    candidate = self.bind_optional_dataclass(data, clazz)
                    score = self.context.class_type.score_object(candidate)
                    if score > max_score:
                        max_score = score
                        obj = candidate

            if obj:
                return obj

            raise ParserError(
                f"Failed to bind object with properties({list(data.keys())}) "
                f"to any of the {[cls.__qualname__ for cls in classes]}"
            )

        def bind_optional_dataclass(self, data: Dict, clazz: Type[T]) -> Optional[T]:
            """Recursively build the given model from the input dict data but fail
            on any converter warnings."""
            try:
                with warnings.catch_warnings():
                    warnings.filterwarnings("error", category=ConverterWarning)
                    return self.bind_dataclass(data, clazz)
            except Exception:
                return None

        def bind_value(
            self, meta: XmlMeta, var: XmlVar, value: Any, recursive: bool = False
        ) -> Any:
            """Main entry point for binding values."""

            # xs:anyAttributes get it out of the way, it's the mapping exception!
            if var.is_attributes:
                return dict(value)

            # Repeating element, recursively bind the values
            if not recursive and var.list_element and isinstance(value, list):
                assert var.factory is not None
                return var.factory(self.bind_value(meta, var, val, True) for val in value)

            # If not dict this is an text or tokens value.
            if not isinstance(value, dict):
                return self.bind_text(meta, var, value)

            keys = value.keys()
            if keys == self.context.class_type.any_keys:
                # Bind data to AnyElement dataclass
                return self.bind_dataclass(value, self.context.class_type.any_element)

            if keys == self.context.class_type.derived_keys:
                # Bind data to AnyElement dataclass
                return self.bind_derived_value(meta, var, value)

            # Bind data to a user defined dataclass
            return self.bind_complex_type(meta, var, value)

        def bind_text(self, meta: XmlMeta, var: XmlVar, value: Any) -> Any:
            """Bind text/tokens value entrypoint."""
            if var.is_elements:
                # Compound field we need to match the value to one of the choice elements
                check_subclass = self.context.class_type.is_model(value)
                choice = var.find_value_choice(value, check_subclass)
                if choice:
                    return self.bind_text(meta, choice, value)

                if value is None:
                    return value

                raise ParserError(
                    f"Failed to bind '{value}' "
                    f"to {meta.clazz.__qualname__}.{var.name} field"
                )

            if var.any_type or var.is_wildcard:
                # field can support any object return the value as it is
                return value

            value = converter.serialize(value)

            # Convert value according to the field types
            return ParserUtils.parse_value(
                value=value,
                types=var.types,
                default=var.default,
                ns_map=EMPTY_MAP,
                tokens_factory=var.tokens_factory,
                format=var.format,
            )

        def bind_complex_type(self, meta: XmlMeta, var: XmlVar, data: Dict) -> Any:
            """Bind data to a user defined dataclass."""
            if var.is_clazz_union:
                # Union of dataclasses
                return self.bind_best_dataclass(data, var.types)
            if var.elements:
                # Compound field with multiple choices
                return self.bind_best_dataclass(data, var.element_types)
            if var.any_type or var.is_wildcard:
                # xs:anyType element, check all meta classes
                return self.bind_best_dataclass(data, meta.element_types)

            assert var.clazz is not None

            subclasses = set(self.context.get_subclasses(var.clazz))
            if subclasses:
                # field annotation is an abstract/base type
                subclasses.add(var.clazz)
                return self.bind_best_dataclass(data, subclasses)

            return self.bind_dataclass(data, var.clazz)

        def bind_derived_value(self, meta: XmlMeta, var: XmlVar, data: Dict) -> Any:
            """Bind derived element entry point."""
            qname = data["qname"]
            xsi_type = data["type"]
            params = data["value"]

            if var.elements:
                choice = var.find_choice(qname)
                if choice is None:
                    raise ParserError(
                        f"Unable to locate compound element"
                        f" {meta.clazz.__qualname__}.{var.name}[{qname}]"
                    )
                return self.bind_derived_value(meta, choice, data)

            if not isinstance(params, dict):
                value = self.bind_text(meta, var, params)
            elif xsi_type:
                clazz: Optional[Type] = self.context.find_type(xsi_type)

                if clazz is None:
                    raise ParserError(f"Unable to locate xsi:type `{xsi_type}`")

                value = self.bind_dataclass(params, clazz)
            elif var.clazz:
                value = self.bind_complex_type(meta, var, params)
            else:
                value = self.bind_best_dataclass(params, meta.element_types)

            generic = self.context.class_type.derived_element
            return generic(qname=qname, value=value, type=xsi_type)

        @classmethod
        def find_var(
            cls, xml_vars: Sequence[XmlVar], local_name: str, is_list: bool = False
        ) -> Optional[XmlVar]:
            for var in xml_vars:
                if var.local_name == local_name:
                    var_is_list = var.list_element or var.tokens
                    if is_list == var_is_list or var.clazz is None:
                        return var

            return None

    @dataclass
    class DictConverter(JsonParser):
        def convert(self, data: Dict, clazz: Type[T]) -> T:
            return self.bind_dataclass(data, clazz)