serde.de

   1"""
   2This module provides `deserialize`, `is_deserializable` `from_dict`, `from_tuple` and classes
   3and functions associated with deserialization.
   4"""
   5
   6from __future__ import annotations
   7import abc
   8import itertools
   9import collections
  10import dataclasses
  11import functools
  12import typing
  13import jinja2
  14from collections.abc import Callable, Sequence, Iterable
  15from beartype import beartype, BeartypeConf
  16from beartype.door import is_bearable
  17from beartype.roar import BeartypeCallHintParamViolation
  18from dataclasses import dataclass, is_dataclass
  19from typing import overload, TypeVar, Generic, Any, Optional, Union, Literal, Iterator
  20from typing_extensions import dataclass_transform
  21
  22from .compat import (
  23    SerdeError,
  24    SerdeSkip,
  25    T,
  26    UserError,
  27    find_generic_arg,
  28    get_args,
  29    get_generic_arg,
  30    get_origin,
  31    get_type_var_names,
  32    is_any,
  33    is_bare_dict,
  34    is_bare_list,
  35    is_bare_set,
  36    is_bare_tuple,
  37    is_datetime,
  38    is_default_dict,
  39    is_dict,
  40    is_ellipsis,
  41    is_enum,
  42    is_frozen_set,
  43    is_generic,
  44    is_list,
  45    is_literal,
  46    is_none,
  47    is_opt,
  48    is_primitive,
  49    is_primitive_subclass,
  50    is_set,
  51    is_str_serializable,
  52    is_tuple,
  53    is_union,
  54    is_variable_tuple,
  55    is_pep695_type_alias,
  56    iter_literals,
  57    iter_types,
  58    iter_unions,
  59    type_args,
  60    typename,
  61)
  62from .core import (
  63    GLOBAL_CLASS_DESERIALIZER,
  64    ClassDeserializer,
  65    FROM_DICT,
  66    FROM_ITER,
  67    SERDE_SCOPE,
  68    CACHE,
  69    UNION_DE_PREFIX,
  70    DefaultTagging,
  71    Field,
  72    disabled,
  73    Scope,
  74    Tagging,
  75    TypeCheck,
  76    add_func,
  77    coerce_object,
  78    strict,
  79    has_default,
  80    has_default_factory,
  81    ensure,
  82    fields,
  83    is_instance,
  84    literal_func_name,
  85    logger,
  86    raise_unsupported_type,
  87    union_func_name,
  88)
  89from .numpy import (
  90    deserialize_numpy_array,
  91    deserialize_numpy_scalar,
  92    deserialize_numpy_array_direct,
  93    deserialize_numpy_jaxtyping_array,
  94    is_numpy_array,
  95    is_numpy_jaxtyping,
  96    is_numpy_scalar,
  97)
  98
  99__all__ = ["deserialize", "is_deserializable", "from_dict", "from_tuple"]
 100
 101
 102DeserializeFunc = Callable[[type[Any], Any], Any]
 103""" Interface of Custom deserialize function. """
 104
 105
 106def serde_legacy_custom_class_deserializer(
 107    cls: type[Any], datavar: Any, value: Any, custom: DeserializeFunc, default: Callable[[], Any]
 108) -> Any:
 109    """
 110    Handle custom deserialization. Use default deserialization logic if it receives `SerdeSkip`
 111    exception.
 112
 113    :param cls: Type of the field.
 114    :param datavar: The whole variable to deserialize from. e.g. "data"
 115    :param value: The value for the field. e.g. "data['i']"
 116    :param custom: Custom deserialize function.
 117    :param default: Default deserialize function.
 118    """
 119    try:
 120        return custom(cls, value)
 121    except SerdeSkip:
 122        return default()
 123
 124
 125def default_deserializer(_cls: type[Any], obj: Any) -> Any:
 126    """
 127    Marker function to tell serde to use the default deserializer. It's used when custom
 128    deserializer is specified at the class but you want to override a field with the default
 129    deserializer.
 130    """
 131
 132
 133def _get_by_aliases(
 134    d: dict[str, str], aliases: list[str], raise_error: bool = True
 135) -> Optional[str]:
 136    if not aliases:
 137        if raise_error:
 138            raise KeyError("Tried all aliases, but key not found")
 139        else:
 140            return None
 141    if aliases[0] in d:
 142        return d[aliases[0]]
 143    else:
 144        return _get_by_aliases(d, aliases[1:], raise_error=raise_error)
 145
 146
 147def _exists_by_aliases(d: dict[str, str], aliases: list[str]) -> bool:
 148    for alias in aliases:
 149        if alias in d:
 150            return True
 151    return False
 152
 153
 154def _make_deserialize(
 155    cls_name: str,
 156    fields: list[Any],
 157    *args: Any,
 158    rename_all: Optional[str] = None,
 159    reuse_instances_default: bool = True,
 160    convert_sets_default: bool = False,
 161    deserializer: Optional[DeserializeFunc] = None,
 162    type_check: TypeCheck = strict,
 163    class_deserializer: Optional[ClassDeserializer] = None,
 164    **kwargs: Any,
 165) -> type[Any]:
 166    """
 167    Create a deserializable class programatically.
 168    """
 169    C: type[Any] = dataclasses.make_dataclass(cls_name, fields, *args, **kwargs)
 170    C = deserialize(
 171        C,
 172        rename_all=rename_all,
 173        reuse_instances_default=reuse_instances_default,
 174        convert_sets_default=convert_sets_default,
 175        **kwargs,
 176    )
 177    return C
 178
 179
 180# The `deserialize` function can call itself recursively when it needs to generate code for
 181# unmarked dataclasses. To avoid infinite recursion, this array remembers types for which code is
 182# currently being generated.
 183GENERATION_STACK = []
 184
 185
 186@dataclass_transform()
 187def deserialize(
 188    _cls: Optional[type[T]] = None,
 189    rename_all: Optional[str] = None,
 190    reuse_instances_default: bool = True,
 191    convert_sets_default: bool = False,
 192    deserializer: Optional[DeserializeFunc] = None,
 193    tagging: Tagging = DefaultTagging,
 194    type_check: TypeCheck = strict,
 195    class_deserializer: Optional[ClassDeserializer] = None,
 196    deny_unknown_fields: bool = False,
 197    **kwargs: Any,
 198) -> type[T]:
 199    """
 200    A dataclass with this decorator is deserializable from any of the data formats supported
 201    by pyserde.
 202
 203    >>> from serde import deserialize
 204    >>> from serde.json import from_json
 205    >>>
 206    >>> @deserialize
 207    ... class Foo:
 208    ...     i: int
 209    ...     s: str
 210    ...     f: float
 211    ...     b: bool
 212    >>>
 213    >>> from_json(Foo, '{"i": 10, "s": "foo", "f": 100.0, "b": true}')
 214    Foo(i=10, s='foo', f=100.0, b=True)
 215    """
 216
 217    stack = []
 218
 219    def wrap(cls: type[T]) -> type[T]:
 220        if cls in stack:
 221            return cls
 222        stack.append(cls)
 223
 224        tagging.check()
 225
 226        # If no `dataclass` found in the class, dataclassify it automatically.
 227        if not is_dataclass(cls):
 228            dataclass(cls)
 229
 230        if type_check.is_strict():
 231            serde_beartype = beartype(conf=BeartypeConf(violation_type=SerdeError))
 232            serde_beartype(cls)
 233
 234        g: dict[str, Any] = {}
 235
 236        # Create a scope storage used by serde.
 237        # Each class should get own scope. Child classes can not share scope with parent class.
 238        # That's why we need the "scope.cls is not cls" check.
 239        scope: Optional[Scope] = getattr(cls, SERDE_SCOPE, None)
 240        if scope is None or scope.cls is not cls:
 241            scope = Scope(cls, reuse_instances_default=reuse_instances_default)
 242            setattr(cls, SERDE_SCOPE, scope)
 243
 244        class_deserializers: list[ClassDeserializer] = list(
 245            itertools.chain(
 246                GLOBAL_CLASS_DESERIALIZER, [class_deserializer] if class_deserializer else []
 247            )
 248        )
 249
 250        # Set some globals for all generated functions
 251        g["cls"] = cls
 252        g["serde_scope"] = scope
 253        g["SerdeError"] = SerdeError
 254        g["UserError"] = UserError
 255        g["raise_unsupported_type"] = raise_unsupported_type
 256        g["typename"] = typename
 257        g["ensure"] = ensure
 258        g["typing"] = typing
 259        g["collections"] = collections
 260        g["Literal"] = Literal
 261        g["from_obj"] = from_obj
 262        g["get_generic_arg"] = get_generic_arg
 263        g["is_instance"] = is_instance
 264        g["TypeCheck"] = TypeCheck
 265        g["disabled"] = disabled
 266        g["coerce_object"] = coerce_object
 267        g["_exists_by_aliases"] = _exists_by_aliases
 268        g["_get_by_aliases"] = _get_by_aliases
 269        g["class_deserializers"] = class_deserializers
 270        g["BeartypeCallHintParamViolation"] = BeartypeCallHintParamViolation
 271        g["is_bearable"] = is_bearable
 272        if deserializer:
 273            g["serde_legacy_custom_class_deserializer"] = functools.partial(
 274                serde_legacy_custom_class_deserializer, custom=deserializer
 275            )
 276
 277        # Collect types used in the generated code.
 278        for typ in iter_types(cls):
 279            # When we encounter a dataclass not marked with deserialize, then also generate
 280            # deserialize functions for it.
 281            if is_dataclass_without_de(typ) and typ is not cls:
 282                # We call deserialize and not wrap to make sure that we will use the default serde
 283                # configuration for generating the deserialization function.
 284                deserialize(typ)
 285
 286            # We don't want to add primitive class e.g "str" into the scope, but primitive
 287            # compatible types such as IntEnum and a subclass of primitives are added,
 288            # so that generated code can use those types.
 289            if is_primitive(typ) and not is_enum(typ) and not is_primitive_subclass(typ):
 290                continue
 291
 292            if is_generic(typ):
 293                g[typename(typ)] = get_origin(typ)
 294            else:
 295                g[typename(typ)] = typ
 296
 297        # render all union functions
 298        for union in iter_unions(cls):
 299            union_args = type_args(union)
 300            add_func(
 301                scope,
 302                union_func_name(UNION_DE_PREFIX, union_args),
 303                render_union_func(cls, union_args, tagging),
 304                g,
 305            )
 306
 307        # render literal functions
 308        for literal in iter_literals(cls):
 309            literal_args = type_args(literal)
 310            add_func(
 311                scope, literal_func_name(literal_args), render_literal_func(cls, literal_args), g
 312            )
 313
 314        # Collect default values and default factories used in the generated code.
 315        for f in defields(cls):
 316            assert f.name
 317            if has_default(f):
 318                scope.defaults[f.name] = f.default
 319            elif has_default_factory(f):
 320                scope.defaults[f.name] = f.default_factory
 321            if f.deserializer:
 322                g[f.deserializer.name] = f.deserializer
 323
 324        add_func(
 325            scope,
 326            FROM_ITER,
 327            render_from_iter(cls, deserializer, type_check, class_deserializer=class_deserializer),
 328            g,
 329        )
 330        add_func(
 331            scope,
 332            FROM_DICT,
 333            render_from_dict(
 334                cls,
 335                rename_all,
 336                deserializer,
 337                type_check,
 338                class_deserializer=class_deserializer,
 339                deny_unknown_fields=deny_unknown_fields,
 340            ),
 341            g,
 342        )
 343
 344        logger.debug(f"{typename(cls)}: {SERDE_SCOPE} {scope}")
 345
 346        stack.pop()
 347        return cls
 348
 349    if _cls is None:
 350        return wrap  # type: ignore
 351
 352    if _cls in GENERATION_STACK:
 353        return _cls
 354
 355    GENERATION_STACK.append(_cls)
 356    try:
 357        return wrap(_cls)
 358    finally:
 359        GENERATION_STACK.pop()
 360
 361
 362def is_deserializable(instance_or_class: Any) -> bool:
 363    """
 364    Test if an instance or class is deserializable.
 365
 366    >>> @deserialize
 367    ... class Foo:
 368    ...     pass
 369    >>>
 370    >>> is_deserializable(Foo)
 371    True
 372    """
 373    return hasattr(instance_or_class, SERDE_SCOPE)
 374
 375
 376def is_dataclass_without_de(cls: type[Any]) -> bool:
 377    if not dataclasses.is_dataclass(cls):
 378        return False
 379    if not hasattr(cls, SERDE_SCOPE):
 380        return True
 381    scope: Optional[Scope] = getattr(cls, SERDE_SCOPE)
 382    if not scope:
 383        return True
 384    return FROM_DICT not in scope.funcs
 385
 386
 387class Deserializer(Generic[T], metaclass=abc.ABCMeta):
 388    """
 389    `Deserializer` base class. Subclass this to customize deserialize behaviour.
 390
 391    See `serde.json.JsonDeserializer` and `serde.msgpack.MsgPackDeserializer` for example usage.
 392    """
 393
 394    @classmethod
 395    @abc.abstractmethod
 396    def deserialize(cls, data: T, **opts: Any) -> Any:
 397        """
 398        deserialize `data` into an object typically `dict`, `list` or `tuple`.
 399
 400        For example, `serde.json.JsonDeserializer` takes json string and deserialize
 401        into an object. `serde.msgpack.MsgPackDeserializer` takes msgpack bytes and
 402        deserialize into an object.
 403        """
 404        raise NotImplementedError
 405
 406
 407def from_obj(c: type[T], o: Any, named: bool, reuse_instances: Optional[bool]) -> T:
 408    """
 409    Deserialize from an object into an instance of the type specified as arg `c`.
 410    `c` can be either primitive type, `list`, `tuple`, `dict` or `deserialize` class.
 411    """
 412
 413    res: Any
 414
 415    # It is possible that the parser already generates the correct data type requested
 416    # by the caller. Hence, it should be checked early to avoid doing anymore work.
 417    if type(o) is c:
 418        return o
 419
 420    def deserializable_to_obj(cls: type[T]) -> T:
 421        serde_scope: Scope = getattr(cls, SERDE_SCOPE)
 422        func_name = FROM_DICT if named else FROM_ITER
 423        res = serde_scope.funcs[func_name](
 424            cls, maybe_generic=maybe_generic, data=o, reuse_instances=reuse_instances
 425        )
 426        return res  # type: ignore
 427
 428    if is_union(c) and not is_opt(c):
 429        # If a class in the argument is a non-dataclass class e.g. Union[Foo, Bar],
 430        # pyserde generates a wrapper (de)serializable dataclass on the fly,
 431        # and use it to deserialize into the object.
 432        return CACHE.deserialize_union(c, o)
 433
 434    if is_generic(c):
 435        # Store subscripted generic type such as Foo[Bar] in "maybe_generic",
 436        # and store origin type such as Foo in "c". Since subscripted generics
 437        # are not a subclass of "type", use "c" for type inspection, and pass
 438        # "maybe_generic" in deserialize functions.
 439        maybe_generic = c
 440        c = get_origin(c)  # type: ignore
 441    else:
 442        maybe_generic = c
 443    try:
 444        thisfunc = functools.partial(from_obj, named=named, reuse_instances=reuse_instances)
 445        if is_dataclass_without_de(c):
 446            # Do not automatically implement beartype if dataclass without serde decorator
 447            # is passed, because it is surprising for users
 448            # See https://github.com/yukinarit/pyserde/issues/480
 449            deserialize(c, type_check=disabled)
 450            res = deserializable_to_obj(c)
 451        elif is_deserializable(c):
 452            res = deserializable_to_obj(c)
 453        elif is_opt(c):
 454            if o is None:
 455                res = None
 456            else:
 457                res = thisfunc(type_args(c)[0], o)
 458        elif is_list(c):
 459            if is_bare_list(c):
 460                res = list(o)
 461            else:
 462                res = [thisfunc(type_args(c)[0], e) for e in o]
 463        elif is_set(c):
 464            if is_bare_set(c):
 465                res = set(o)
 466            elif is_frozen_set(c):
 467                res = frozenset(thisfunc(type_args(c)[0], e) for e in o)
 468            else:
 469                res = {thisfunc(type_args(c)[0], e) for e in o}
 470        elif is_tuple(c):
 471            if is_bare_tuple(c) or is_variable_tuple(c):
 472                res = tuple(e for e in o)
 473            else:
 474                res = tuple(thisfunc(type_args(c)[i], e) for i, e in enumerate(o))
 475        elif is_dict(c):
 476            if is_bare_dict(c):
 477                res = o
 478            elif is_default_dict(c):
 479                f = DeField(c, "")
 480                v = f.value_field()
 481                origin = get_origin(v.type)
 482                res = collections.defaultdict(
 483                    origin if origin else v.type,
 484                    {
 485                        thisfunc(type_args(c)[0], k): thisfunc(type_args(c)[1], v)
 486                        for k, v in o.items()
 487                    },
 488                )
 489            else:
 490                res = {
 491                    thisfunc(type_args(c)[0], k): thisfunc(type_args(c)[1], v) for k, v in o.items()
 492                }
 493        elif is_numpy_array(c):
 494            res = deserialize_numpy_array_direct(c, o)
 495        elif is_datetime(c):
 496            res = c.fromisoformat(o)
 497        elif is_any(c) or is_ellipsis(c):
 498            res = o
 499        else:
 500            res = c(o)  # type: ignore
 501
 502        return res  # type: ignore
 503
 504    except UserError as e:
 505        raise e.inner from None
 506
 507    except Exception as e:
 508        raise SerdeError(e) from None
 509
 510
 511@overload
 512def from_dict(cls: type[T], o: dict[str, Any], reuse_instances: Optional[bool] = None) -> T: ...
 513
 514
 515@overload
 516def from_dict(cls: Any, o: dict[str, Any], reuse_instances: Optional[bool] = None) -> Any: ...
 517
 518
 519def from_dict(cls: Any, o: dict[str, Any], reuse_instances: Optional[bool] = None) -> Any:
 520    """
 521    Deserialize dictionary into object.
 522
 523    >>> @deserialize
 524    ... class Foo:
 525    ...     i: int
 526    ...     s: str = 'foo'
 527    ...     f: float = 100.0
 528    ...     b: bool = True
 529    >>>
 530    >>> from_dict(Foo, {'i': 10, 's': 'foo', 'f': 100.0, 'b': True})
 531    Foo(i=10, s='foo', f=100.0, b=True)
 532
 533    You can pass any type supported by pyserde. For example,
 534
 535    >>> lst = [{'i': 10, 's': 'foo', 'f': 100.0, 'b': True},
 536    ...        {'i': 20, 's': 'foo', 'f': 100.0, 'b': True}]
 537    >>> from_dict(list[Foo], lst)
 538    [Foo(i=10, s='foo', f=100.0, b=True), Foo(i=20, s='foo', f=100.0, b=True)]
 539    """
 540    return from_obj(cls, o, named=True, reuse_instances=reuse_instances)
 541
 542
 543@overload
 544def from_tuple(cls: type[T], o: Any, reuse_instances: Optional[bool] = None) -> T: ...
 545
 546
 547@overload
 548def from_tuple(cls: Any, o: Any, reuse_instances: Optional[bool] = None) -> Any: ...
 549
 550
 551def from_tuple(cls: Any, o: Any, reuse_instances: Optional[bool] = None) -> Any:
 552    """
 553    Deserialize tuple into object.
 554
 555    >>> @deserialize
 556    ... class Foo:
 557    ...     i: int
 558    ...     s: str = 'foo'
 559    ...     f: float = 100.0
 560    ...     b: bool = True
 561    >>>
 562    >>> from_tuple(Foo, (10, 'foo', 100.0, True))
 563    Foo(i=10, s='foo', f=100.0, b=True)
 564
 565    You can pass any type supported by pyserde. For example,
 566
 567    >>> lst = [(10, 'foo', 100.0, True), (20, 'foo', 100.0, True)]
 568    >>> from_tuple(list[Foo], lst)
 569    [Foo(i=10, s='foo', f=100.0, b=True), Foo(i=20, s='foo', f=100.0, b=True)]
 570    """
 571    return from_obj(cls, o, named=False, reuse_instances=reuse_instances)
 572
 573
 574@dataclass
 575class DeField(Field[T]):
 576    """
 577    Represents a field of dataclass.
 578    """
 579
 580    datavar: Optional[str] = None
 581    """ Name of variable which is passed in the deserialize API """
 582
 583    index: int = 0
 584    """ Field index """
 585
 586    iterbased: bool = False
 587    """ Iterater based deserializer or not """
 588
 589    def __getitem__(self, n: int) -> Union[DeField[Any], InnerField[Any]]:
 590        """
 591        Get inner `Field` from current `Field`.
 592
 593        `InnerField` is returned if self is of any standard collection e.g. list.
 594        `DeField` is returned if self is Optional.
 595        """
 596        typ = type_args(self.type)[n]
 597        opts: dict[str, Any] = {
 598            "kw_only": self.kw_only,
 599            "case": self.case,
 600            "alias": self.alias,
 601            "rename": self.rename,
 602            "skip": self.skip,
 603            "skip_if": self.skip_if,
 604            "skip_if_false": self.skip_if_false,
 605            "skip_if_default": self.skip_if_default,
 606            "serializer": self.serializer,
 607            "deserializer": self.deserializer,
 608            "flatten": self.flatten,
 609            "parent": self.parent,
 610        }
 611        if is_list(self.type) or is_dict(self.type) or is_set(self.type):
 612            return InnerField(typ, "v", datavar="v", **opts)
 613        elif is_tuple(self.type):
 614            return InnerField(typ, f"{self.data}[{n}]", datavar=f"{self.data}[{n}]", **opts)
 615        else:
 616            # For Optional etc.
 617            return DeField(
 618                typ,
 619                self.name,
 620                datavar=self.datavar,
 621                index=self.index,
 622                iterbased=self.iterbased,
 623                **opts,
 624            )
 625
 626    def key_field(self) -> DeField[Any]:
 627        """
 628        Get inner key field for dict like class.
 629        """
 630        k = self[0]
 631        k.name = "k"
 632        k.datavar = "k"
 633        return k
 634
 635    def value_field(self) -> DeField[Any]:
 636        """
 637        Get inner value field for dict like class.
 638        """
 639        return self[1]
 640
 641    @property
 642    def data(self) -> str:
 643        """
 644        Renders the variable name that possesses the data.
 645
 646        e.g. tuple
 647            * datavar property returns "data"
 648            * data property returns "data[0]".
 649        e.g. Optional
 650            * datavar property returns "data"
 651            * data property returns "data.get("field_name")".
 652        For other types
 653            * datavar property returns "data"
 654            * data property returns "data["field_name"]".
 655        """
 656
 657        if self.iterbased:
 658            return f"{self.datavar}[{self.index}]"
 659        elif is_union(self.type) and type(None) in get_args(self.type):
 660            return f'{self.datavar}.get("{self.conv_name()}")'
 661        else:
 662            return f'{self.datavar}["{self.conv_name()}"]'
 663
 664    @data.setter
 665    def data(self, d: str) -> None:
 666        self.datavar = d
 667
 668    def data_or(self) -> str:
 669        if self.iterbased:
 670            return self.data
 671        else:
 672            return f'{self.datavar}.get("{self.conv_name()}")'
 673
 674
 675@dataclass
 676class InnerField(DeField[T]):
 677    """
 678    Field of Inner type. The purpose of this class is to override "data" method
 679    for inner type codegen.
 680
 681    e.g.
 682      T of list[T]
 683      V of dict[K, V]
 684      T of Optional[T]
 685    """
 686
 687    @property
 688    def data(self) -> str:
 689        return self.datavar or ""
 690
 691    @data.setter
 692    def data(self, d: str) -> None:
 693        self.datavar = d
 694
 695
 696def defields(cls: type[Any]) -> list[DeField[Any]]:
 697    return fields(DeField, cls)
 698
 699
 700@dataclass
 701class Renderer:
 702    """
 703    Render rvalue for code generation.
 704    """
 705
 706    func: str
 707    cls: Optional[type[Any]] = None
 708    legacy_class_deserializer: Optional[DeserializeFunc] = None
 709    import_numpy: bool = False
 710    suppress_coerce: bool = False
 711    """ Disable type coercing in codegen """
 712    class_deserializer: Optional[ClassDeserializer] = None
 713    class_name: Optional[str] = None
 714
 715    def render(self, arg: DeField[Any]) -> str:
 716        """
 717        Render rvalue
 718        """
 719        implemented_methods: dict[type[Any], int] = {}
 720        class_deserializers: Iterable[ClassDeserializer] = itertools.chain(
 721            GLOBAL_CLASS_DESERIALIZER, [self.class_deserializer] if self.class_deserializer else []
 722        )
 723        for n, class_deserializer in enumerate(class_deserializers):
 724            for method in class_deserializer.__class__.deserialize.methods:  # type: ignore
 725                implemented_methods[get_args(method.signature.types[1])[0]] = n
 726
 727        custom_deserializer_available = arg.type in implemented_methods
 728        if custom_deserializer_available and not arg.deserializer:
 729            res = (
 730                f"class_deserializers[{implemented_methods[arg.type]}].deserialize("
 731                f"{typename(arg.type)}, {arg.data})"
 732            )
 733        elif arg.deserializer and arg.deserializer.inner is not default_deserializer:
 734            res = self.custom_field_deserializer(arg)
 735        elif is_generic(arg.type):
 736            arg.type_args = list(get_args(arg.type))
 737            origin = get_origin(arg.type)
 738            assert origin
 739            arg.type = origin
 740            res = self.render(arg)
 741        elif is_dataclass(arg.type):
 742            res = self.dataclass(arg)
 743        elif is_opt(arg.type):
 744            res = self.opt(arg)
 745        elif is_list(arg.type):
 746            res = self.list(arg)
 747        elif is_set(arg.type):
 748            res = self.set(arg)
 749        elif is_dict(arg.type):
 750            res = self.dict(arg)
 751        elif is_tuple(arg.type):
 752            res = self.tuple(arg)
 753        elif is_enum(arg.type):
 754            res = self.enum(arg)
 755        elif is_numpy_scalar(arg.type):
 756            self.import_numpy = True
 757            res = deserialize_numpy_scalar(arg)
 758        elif is_numpy_array(arg.type):
 759            self.import_numpy = True
 760            res = deserialize_numpy_array(arg)
 761        elif is_numpy_jaxtyping(arg.type):
 762            self.import_numpy = True
 763            res = deserialize_numpy_jaxtyping_array(arg)
 764        elif is_union(arg.type):
 765            res = self.union_func(arg)
 766        elif is_str_serializable(arg.type):
 767            res = f"({self.c_tor_with_check(arg)}) if reuse_instances else {self.c_tor(arg)}"
 768        elif is_datetime(arg.type):
 769            from_iso = f"{typename(arg.type)}.fromisoformat({arg.data})"
 770            res = f"({arg.data} if isinstance({arg.data}, {typename(arg.type)}) else {from_iso}) \
 771                    if reuse_instances else {from_iso}"
 772        elif is_none(arg.type):
 773            res = "None"
 774        elif is_any(arg.type) or is_ellipsis(arg.type):
 775            res = arg.data
 776        elif is_pep695_type_alias(arg.type):
 777            res = self.render(DeField(name=arg.name, type=arg.type.__value__, datavar=arg.datavar))
 778        elif is_primitive(arg.type):
 779            # For subclasses for primitives e.g. class FooStr(str), coercing is always enabled
 780            res = self.primitive(arg, not is_primitive_subclass(arg.type))
 781        elif isinstance(arg.type, TypeVar):
 782            if not self.cls:
 783                raise SerdeError("Missing cls")
 784            index = find_generic_arg(self.cls, arg.type)
 785            res = (
 786                f"from_obj(get_generic_arg(maybe_generic, maybe_generic_type_vars, "
 787                f"variable_type_args, {index}), {arg.data}, named={not arg.iterbased}, "
 788                "reuse_instances=reuse_instances)"
 789            )
 790        elif is_literal(arg.type):
 791            res = self.literal(arg)
 792        else:
 793            return f"raise_unsupported_type({arg.data})"
 794
 795        if arg.supports_default():
 796            res = self.default(arg, res)
 797
 798        if (
 799            self.legacy_class_deserializer
 800            and not arg.deserializer
 801            and not custom_deserializer_available
 802        ):
 803            # Rerender the code for default deserializer.
 804            default = Renderer(
 805                self.func, self.cls, None, suppress_coerce=self.suppress_coerce
 806            ).render(arg)
 807            return self.custom_class_deserializer(arg, default)
 808        else:
 809            return res
 810
 811    def custom_field_deserializer(self, arg: DeField[Any]) -> str:
 812        """
 813        Render rvalue for the field with custom deserializer.
 814        """
 815        if not arg.deserializer:
 816            raise SerdeError("Missing custom field deserializer")
 817        return f"{arg.deserializer.name}({arg.data})"
 818
 819    def custom_class_deserializer(self, arg: DeField[Any], code: str) -> str:
 820        """
 821        Render custom class deserializer.
 822        """
 823        # The function takes a closure in order to execute the default value lazily.
 824        return (
 825            "serde_legacy_custom_class_deserializer("
 826            f"{typename(arg.type)}, "
 827            f"{arg.datavar}, "
 828            f"{arg.data_or()}, "
 829            f"default=lambda: {code})"
 830        )
 831
 832    def dataclass(self, arg: DeField[Any]) -> str:
 833        if not arg.flatten:
 834            # e.g. "data['field']" will be used as variable name.
 835            var = arg.data
 836        else:
 837            # Because the field is flattened
 838            # e.g. "data" will be used as variable name.
 839            assert arg.datavar
 840            if arg.iterbased:
 841                var = f"{arg.datavar}[{arg.index}:]"
 842            else:
 843                var = arg.datavar
 844
 845        type_args_str = [str(t).lstrip("~") for t in arg.type_args] if arg.type_args else None
 846
 847        opts = (
 848            "maybe_generic=maybe_generic, maybe_generic_type_vars=maybe_generic_type_vars, "
 849            f"variable_type_args={type_args_str}, reuse_instances=reuse_instances"
 850        )
 851
 852        if arg.is_self_referencing():
 853            class_name = "cls"
 854        else:
 855            class_name = typename(arg.type)
 856
 857        return f"{class_name}.{SERDE_SCOPE}.funcs['{self.func}'](data={var}, {opts})"
 858
 859    def opt(self, arg: DeField[Any]) -> str:
 860        """
 861        Render rvalue for Optional.
 862        """
 863        inner = arg[0]
 864        if arg.iterbased:
 865            exists = f"{arg.data} is not None"
 866        elif arg.flatten:
 867            # Check nullabilities of all nested fields.
 868            exists = " and ".join(
 869                [
 870                    f'{arg.datavar}.get("{f.name}") is not None'
 871                    for f in dataclasses.fields(inner.type)
 872                ]
 873            )
 874        else:
 875            name = arg.conv_name()
 876            if arg.alias:
 877                aliases = (f'"{s}"' for s in [name, *arg.alias])
 878                get = f"_get_by_aliases(data, [{','.join(aliases)}], raise_error=False)"
 879            else:
 880                get = f'{arg.datavar}.get("{name}")'
 881            exists = f"{get} is not None"
 882        return f"({self.render(inner)}) if {exists} else None"
 883
 884    def list(self, arg: DeField[Any]) -> str:
 885        """
 886        Render rvalue for list.
 887        """
 888        if is_bare_list(arg.type):
 889            return f"list({arg.data})"
 890        else:
 891            return f"[{self.render(arg[0])} for v in {arg.data}]"
 892
 893    def set(self, arg: DeField[Any]) -> str:
 894        """
 895        Render rvalue for set.
 896        """
 897        if is_bare_set(arg.type):
 898            return f"set({arg.data})"
 899        elif is_frozen_set(arg.type):
 900            return f"frozenset({self.render(arg[0])} for v in {arg.data})"
 901        else:
 902            return f"set({self.render(arg[0])} for v in {arg.data})"
 903
 904    def tuple(self, arg: DeField[Any]) -> str:
 905        """
 906        Render rvalue for tuple.
 907        """
 908        if is_bare_tuple(arg.type):
 909            return f"tuple({arg.data})"
 910        elif is_variable_tuple(arg.type):
 911            earg = arg[0]
 912            earg.datavar = "v"
 913            return f"tuple({self.render(earg)} for v in {arg.data})"
 914        else:
 915            values = []
 916            for i, _typ in enumerate(type_args(arg.type)):
 917                inner = arg[i]
 918                values.append(self.render(inner))
 919            return f'({", ".join(values)},)'  # trailing , is required for single element tuples
 920
 921    def dict(self, arg: DeField[Any]) -> str:
 922        """
 923        Render rvalue for dict.
 924        """
 925        if is_bare_dict(arg.type):
 926            return arg.data
 927        elif is_default_dict(arg.type):
 928            k = arg.key_field()
 929            v = arg.value_field()
 930            origin = get_origin(v.type)
 931            if origin:
 932                # When the callable type is of generic type e.g list.
 933                # Get origin type "list" from "list[X]".
 934                callable = origin.__name__
 935            else:
 936                # When the callable type is non generic type e.g int, Foo.
 937                callable = v.type.__name__
 938            return f"collections.defaultdict({callable}, \
 939                    {{{self.render(k)}: {self.render(v)} for k, v in {arg.data}.items()}})"
 940        else:
 941            k = arg.key_field()
 942            v = arg.value_field()
 943            return f"{{{self.render(k)}: {self.render(v)} for k, v in {arg.data}.items()}}"
 944
 945    def enum(self, arg: DeField[Any]) -> str:
 946        return f"{typename(arg.type)}({self.primitive(arg)})"
 947
 948    def primitive(self, arg: DeField[Any], suppress_coerce: bool = False) -> str:
 949        """
 950        Render rvalue for primitives.
 951
 952        * `suppress_coerce`: Overrides "suppress_coerce" in the Renderer's field
 953        """
 954        typ = typename(arg.type)
 955        dat = arg.data
 956        if arg.alias:
 957            aliases = (f'"{s}"' for s in [arg.name, *arg.alias])
 958            dat = f"_get_by_aliases(data, [{','.join(aliases)}])"
 959        if self.suppress_coerce and suppress_coerce:
 960            return dat
 961        else:
 962            assert arg.name
 963            escaped_arg_name = arg.name.replace('"', '\\"')
 964            return f'coerce_object("{self.class_name}", "{escaped_arg_name}", {typ}, {dat})'
 965
 966    def c_tor(self, arg: DeField[Any]) -> str:
 967        return f"{typename(arg.type)}({arg.data})"
 968
 969    def c_tor_with_check(self, arg: DeField[Any], ctor: Optional[str] = None) -> str:
 970        if ctor is None:
 971            ctor = self.c_tor(arg)
 972        return f"{arg.data} if isinstance({arg.data}, {typename(arg.type)}) else {ctor}"
 973
 974    def union_func(self, arg: DeField[Any]) -> str:
 975        func_name = union_func_name(UNION_DE_PREFIX, type_args(arg.type))
 976        return (
 977            f"serde_scope.funcs['{func_name}']("
 978            "cls=cls, "
 979            f"data={arg.data}, "
 980            "reuse_instances=reuse_instances)"
 981        )
 982
 983    def literal(self, arg: DeField[Any]) -> str:
 984        func_name = literal_func_name(type_args(arg.type))
 985        return (
 986            f"serde_scope.funcs['{func_name}']("
 987            "cls=cls, "
 988            f"data={arg.data}, "
 989            "reuse_instances=reuse_instances)"
 990        )
 991
 992    def default(self, arg: DeField[Any], code: str) -> str:
 993        """
 994        Renders supplying default value during deserialization.
 995        """
 996
 997        def get_aliased_fields(arg: Field[Any]) -> Iterator[str]:
 998            return (f'"{s}"' for s in [arg.name, *arg.alias])
 999
1000        if arg.flatten:
1001            # When a field has the `flatten` attribute, iterate over its dataclass fields.
1002            # This ensures that the code checks keys in the data while considering aliases.
1003            flattened = []
1004            for subarg in defields(arg.type):
1005                if subarg.alias:
1006                    aliases = get_aliased_fields(subarg)
1007                    flattened.append(f'_exists_by_aliases({arg.datavar}, [{",".join(aliases)}])')
1008                else:
1009                    flattened.append(f'"{subarg.name}" in {arg.datavar}')
1010            exists = " and ".join(flattened)
1011        else:
1012            if arg.alias:
1013                aliases = get_aliased_fields(arg)
1014                exists = f'_exists_by_aliases({arg.datavar}, [{",".join(aliases)}])'
1015            else:
1016                exists = f'"{arg.conv_name()}" in {arg.datavar}'
1017
1018        if has_default(arg):
1019            return f'({code}) if {exists} else serde_scope.defaults["{arg.name}"]'
1020        elif has_default_factory(arg):
1021            return f'({code}) if {exists} else serde_scope.defaults["{arg.name}"]()'
1022        else:
1023            return code
1024
1025
1026def to_arg(f: DeField[T], index: int, rename_all: Optional[str] = None) -> DeField[T]:
1027    f.index = index
1028    f.data = "data"
1029    f.case = f.case or rename_all
1030    return f
1031
1032
1033def to_iter_arg(f: DeField[T], *args: Any, **kwargs: Any) -> DeField[T]:
1034    f = to_arg(f, *args, **kwargs)
1035    f.iterbased = True
1036    return f
1037
1038
1039def renderable(f: DeField[Any]) -> bool:
1040    return f.init
1041
1042
1043jinja2_env = jinja2.Environment(
1044    loader=jinja2.DictLoader(
1045        {
1046            "iter": """
1047def {{func}}(cls=cls, maybe_generic=None, maybe_generic_type_vars=None, data=None,
1048             variable_type_args=None, reuse_instances=None):
1049  if reuse_instances is None:
1050    reuse_instances = {{serde_scope.reuse_instances_default}}
1051
1052  maybe_generic_type_vars = maybe_generic_type_vars or {{cls_type_vars}}
1053
1054  {% for f in fields %}
1055  __{{f.name}} = {{rvalue(arg(f,loop.index-1))}}
1056  {% endfor %}
1057
1058  try:
1059    return cls(
1060      {% for f in fields %}
1061      __{{f.name}},
1062      {% endfor %}
1063    )
1064  except BeartypeCallHintParamViolation as e:
1065    raise SerdeError(e)
1066  except Exception as e:
1067    raise UserError(e)
1068""",
1069            "dict": """
1070def {{func}}(cls=cls, maybe_generic=None, maybe_generic_type_vars=None, data=None,
1071             variable_type_args=None, reuse_instances=None):
1072  if reuse_instances is None:
1073    reuse_instances = {{serde_scope.reuse_instances_default}}
1074
1075  {% if deny_unknown_fields %}
1076  known_fields = {{ known_fields }}
1077  unknown_fields = set((data or {}).keys()) - known_fields
1078  if unknown_fields:
1079    raise SerdeError(f'unknown fields: {unknown_fields}, expected one of {known_fields}')
1080  {% endif %}
1081
1082  maybe_generic_type_vars = maybe_generic_type_vars or {{cls_type_vars}}
1083
1084  {% for f in fields %}
1085  __{{f.name}} = {{rvalue(arg(f,loop.index-1))}}
1086  {% endfor %}
1087
1088  try:
1089    return cls(
1090    {% for f in fields %}
1091    {% if f.kw_only %}
1092    {{f.name}}=__{{f.name}},
1093    {% else %}
1094    __{{f.name}},
1095    {% endif %}
1096    {% endfor %}
1097    )
1098  except BeartypeCallHintParamViolation as e:
1099    raise SerdeError(e)
1100  except Exception as e:
1101    raise UserError(e)
1102""",
1103            "union": """
1104def {{func}}(cls=cls, maybe_generic=None, maybe_generic_type_vars=None, data=None,
1105             variable_type_args=None, reuse_instances = {{serde_scope.reuse_instances_default}}):
1106  errors = []
1107  {% for t in union_args %}
1108  try:
1109    # create fake dict so we can reuse the normal render function
1110    {% if tagging.is_external() and is_taggable(t)  %}
1111    ensure("{{typename(t)}}" in data , "'{{typename(t)}}' key is not present")
1112    fake_dict = {"fake_key": data["{{typename(t)}}"]}
1113
1114    {% elif tagging.is_internal() and is_taggable(t) %}
1115    ensure("{{tagging.tag}}" in data , "'{{tagging.tag}}' key is not present")
1116    ensure("{{typename(t)}}" == data["{{tagging.tag}}"], "tag '{{typename(t)}}' isn't found")
1117    fake_dict = {"fake_key": data}
1118
1119    {% elif tagging.is_adjacent() and is_taggable(t) %}
1120    ensure("{{tagging.tag}}" in data , "'{{tagging.tag}}' key is not present")
1121    ensure("{{tagging.content}}" in data , "'{{tagging.content}}' key is not present")
1122    ensure("{{typename(t)}}" == data["{{tagging.tag}}"], "tag '{{typename(t)}}' isn't found")
1123    fake_dict = {"fake_key": data["{{tagging.content}}"]}
1124
1125    {% else %}
1126    fake_dict = {"fake_key": data}
1127    {% endif %}
1128
1129    {% if is_primitive(t) or is_none(t) %}
1130    if not isinstance(fake_dict["fake_key"], {{typename(t)}}):
1131        raise Exception("Not a type of {{typename(t)}}")
1132    {% endif %}
1133    res = {{rvalue(arg(t))}}
1134    ensure(is_bearable(res, {{typename(t)}}), "object is not of type '{{typename(t)}}'")
1135    return res
1136  except Exception as e:
1137    errors.append(f" Failed to deserialize into {{typename(t)}}: {e}")
1138  {% endfor %}
1139  raise SerdeError("Can not deserialize " + repr(data) + " of type " + \
1140          typename(type(data)) + " into {{union_name}}.\\nReasons:\\n" + "\\n".join(errors))
1141""",
1142            "literal": """
1143def {{func}}(cls=cls, maybe_generic=None, maybe_generic_type_vars=None, data=None,
1144             variable_type_args=None, reuse_instances = {{serde_scope.reuse_instances_default}}):
1145  if data in ({%- for v in literal_args -%}{{repr(v)}},{%- endfor -%}):
1146    return data
1147  raise SerdeError("Can not deserialize " + repr(data) + " as {{literal_name}}.")
1148  """,
1149        }
1150    )
1151)
1152
1153
1154def render_from_iter(
1155    cls: type[Any],
1156    legacy_class_deserializer: Optional[DeserializeFunc] = None,
1157    type_check: TypeCheck = strict,
1158    class_deserializer: Optional[ClassDeserializer] = None,
1159) -> str:
1160    renderer = Renderer(
1161        FROM_ITER,
1162        cls=cls,
1163        legacy_class_deserializer=legacy_class_deserializer,
1164        suppress_coerce=(not type_check.is_coerce()),
1165        class_deserializer=class_deserializer,
1166        class_name=typename(cls),
1167    )
1168    fields = list(filter(renderable, defields(cls)))
1169    res = jinja2_env.get_template("iter").render(
1170        func=FROM_ITER,
1171        serde_scope=getattr(cls, SERDE_SCOPE),
1172        fields=fields,
1173        cls_type_vars=get_type_var_names(cls),
1174        rvalue=renderer.render,
1175        arg=to_iter_arg,
1176    )
1177
1178    if renderer.import_numpy:
1179        res = "import numpy\n" + res
1180
1181    return res
1182
1183
1184def get_known_fields(f: DeField[Any], rename_all: Optional[str]) -> list[str]:
1185    names: list[str] = [f.conv_name(rename_all)]
1186    return names + f.alias
1187
1188
1189def render_from_dict(
1190    cls: type[Any],
1191    rename_all: Optional[str] = None,
1192    legacy_class_deserializer: Optional[DeserializeFunc] = None,
1193    type_check: TypeCheck = strict,
1194    class_deserializer: Optional[ClassDeserializer] = None,
1195    deny_unknown_fields: bool = False,
1196) -> str:
1197    renderer = Renderer(
1198        FROM_DICT,
1199        cls=cls,
1200        legacy_class_deserializer=legacy_class_deserializer,
1201        suppress_coerce=(not type_check.is_coerce()),
1202        class_deserializer=class_deserializer,
1203        class_name=typename(cls),
1204    )
1205    fields = list(filter(renderable, defields(cls)))
1206    known_fields = set(
1207        itertools.chain.from_iterable([get_known_fields(f, rename_all) for f in fields])
1208    )
1209    res = jinja2_env.get_template("dict").render(
1210        func=FROM_DICT,
1211        serde_scope=getattr(cls, SERDE_SCOPE),
1212        fields=fields,
1213        type_check=type_check,
1214        cls_type_vars=get_type_var_names(cls),
1215        rvalue=renderer.render,
1216        arg=functools.partial(to_arg, rename_all=rename_all),
1217        deny_unknown_fields=deny_unknown_fields,
1218        known_fields=known_fields,
1219    )
1220
1221    if renderer.import_numpy:
1222        res = "import numpy\n" + res
1223
1224    return res
1225
1226
1227def render_union_func(
1228    cls: type[Any], union_args: Sequence[type[Any]], tagging: Tagging = DefaultTagging
1229) -> str:
1230    union_name = f"Union[{', '.join([typename(a) for a in union_args])}]"
1231
1232    renderer = Renderer(FROM_DICT, cls=cls, suppress_coerce=True)
1233    return jinja2_env.get_template("union").render(
1234        func=union_func_name(UNION_DE_PREFIX, union_args),
1235        serde_scope=getattr(cls, SERDE_SCOPE),
1236        union_args=union_args,
1237        union_name=union_name,
1238        tagging=tagging,
1239        is_taggable=Tagging.is_taggable,
1240        arg=lambda x: DeField(x, datavar="fake_dict", name="fake_key"),
1241        rvalue=renderer.render,
1242        is_primitive=is_primitive,
1243        is_none=is_none,
1244        typename=typename,
1245    )
1246
1247
1248def render_literal_func(
1249    cls: type[Any], literal_args: Sequence[Any], tagging: Tagging = DefaultTagging
1250) -> str:
1251    literal_name = f"Literal[{', '.join([repr(a) for a in literal_args])}]"
1252    return jinja2_env.get_template("literal").render(
1253        func=literal_func_name(literal_args),
1254        serde_scope=getattr(cls, SERDE_SCOPE),
1255        literal_args=literal_args,
1256        literal_name=literal_name,
1257        tagging=tagging,
1258        is_taggable=Tagging.is_taggable,
1259        repr=repr,
1260        type=type,
1261    )