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