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    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        """
 989        Renders supplying default value during deserialization.
 990        """
 991
 992        def get_aliased_fields(arg: Field[Any]) -> Iterator[str]:
 993            return (f'"{s}"' for s in [arg.name, *arg.alias])
 994
 995        if arg.flatten:
 996            # When a field has the `flatten` attribute, iterate over its dataclass fields.
 997            # This ensures that the code checks keys in the data while considering aliases.
 998            flattened = []
 999            for subarg in defields(arg.type):
1000                if subarg.alias:
1001                    aliases = get_aliased_fields(subarg)
1002                    flattened.append(f'_exists_by_aliases({arg.datavar}, [{",".join(aliases)}])')
1003                else:
1004                    flattened.append(f'"{subarg.name}" in {arg.datavar}')
1005            exists = " and ".join(flattened)
1006        else:
1007            if arg.alias:
1008                aliases = get_aliased_fields(arg)
1009                exists = f'_exists_by_aliases({arg.datavar}, [{",".join(aliases)}])'
1010            else:
1011                exists = f'"{arg.conv_name()}" in {arg.datavar}'
1012
1013        if has_default(arg):
1014            return f'({code}) if {exists} else serde_scope.defaults["{arg.name}"]'
1015        elif has_default_factory(arg):
1016            return f'({code}) if {exists} else serde_scope.defaults["{arg.name}"]()'
1017        else:
1018            return code
1019
1020
1021def to_arg(f: DeField[T], index: int, rename_all: Optional[str] = None) -> DeField[T]:
1022    f.index = index
1023    f.data = "data"
1024    f.case = f.case or rename_all
1025    return f
1026
1027
1028def to_iter_arg(f: DeField[T], *args: Any, **kwargs: Any) -> DeField[T]:
1029    f = to_arg(f, *args, **kwargs)
1030    f.iterbased = True
1031    return f
1032
1033
1034def renderable(f: DeField[Any]) -> bool:
1035    return f.init
1036
1037
1038jinja2_env = jinja2.Environment(
1039    loader=jinja2.DictLoader(
1040        {
1041            "iter": """
1042def {{func}}(cls=cls, maybe_generic=None, maybe_generic_type_vars=None, data=None,
1043             variable_type_args=None, reuse_instances=None):
1044  if reuse_instances is None:
1045    reuse_instances = {{serde_scope.reuse_instances_default}}
1046
1047  maybe_generic_type_vars = maybe_generic_type_vars or {{cls_type_vars}}
1048
1049  {% for f in fields %}
1050  __{{f.name}} = {{rvalue(arg(f,loop.index-1))}}
1051  {% endfor %}
1052
1053  try:
1054    return cls(
1055      {% for f in fields %}
1056      __{{f.name}},
1057      {% endfor %}
1058    )
1059  except BeartypeCallHintParamViolation as e:
1060    raise SerdeError(e)
1061  except Exception as e:
1062    raise UserError(e)
1063""",
1064            "dict": """
1065def {{func}}(cls=cls, maybe_generic=None, maybe_generic_type_vars=None, data=None,
1066             variable_type_args=None, reuse_instances=None):
1067  if reuse_instances is None:
1068    reuse_instances = {{serde_scope.reuse_instances_default}}
1069
1070  {% if deny_unknown_fields %}
1071  known_fields = {{ known_fields }}
1072  unknown_fields = set((data or {}).keys()) - known_fields
1073  if unknown_fields:
1074    raise SerdeError(f'unknown fields: {unknown_fields}, expected one of {known_fields}')
1075  {% endif %}
1076
1077  maybe_generic_type_vars = maybe_generic_type_vars or {{cls_type_vars}}
1078
1079  {% for f in fields %}
1080  __{{f.name}} = {{rvalue(arg(f,loop.index-1))}}
1081  {% endfor %}
1082
1083  try:
1084    return cls(
1085    {% for f in fields %}
1086    {% if f.kw_only %}
1087    {{f.name}}=__{{f.name}},
1088    {% else %}
1089    __{{f.name}},
1090    {% endif %}
1091    {% endfor %}
1092    )
1093  except BeartypeCallHintParamViolation as e:
1094    raise SerdeError(e)
1095  except Exception as e:
1096    raise UserError(e)
1097""",
1098            "union": """
1099def {{func}}(cls=cls, maybe_generic=None, maybe_generic_type_vars=None, data=None,
1100             variable_type_args=None, reuse_instances = {{serde_scope.reuse_instances_default}}):
1101  errors = []
1102  {% for t in union_args %}
1103  try:
1104    # create fake dict so we can reuse the normal render function
1105    {% if tagging.is_external() and is_taggable(t)  %}
1106    ensure("{{typename(t)}}" in data , "'{{typename(t)}}' key is not present")
1107    fake_dict = {"fake_key": data["{{typename(t)}}"]}
1108
1109    {% elif tagging.is_internal() and is_taggable(t) %}
1110    ensure("{{tagging.tag}}" in data , "'{{tagging.tag}}' key is not present")
1111    ensure("{{typename(t)}}" == data["{{tagging.tag}}"], "tag '{{typename(t)}}' isn't found")
1112    fake_dict = {"fake_key": data}
1113
1114    {% elif tagging.is_adjacent() and is_taggable(t) %}
1115    ensure("{{tagging.tag}}" in data , "'{{tagging.tag}}' key is not present")
1116    ensure("{{tagging.content}}" in data , "'{{tagging.content}}' key is not present")
1117    ensure("{{typename(t)}}" == data["{{tagging.tag}}"], "tag '{{typename(t)}}' isn't found")
1118    fake_dict = {"fake_key": data["{{tagging.content}}"]}
1119
1120    {% else %}
1121    fake_dict = {"fake_key": data}
1122    {% endif %}
1123
1124    {% if is_primitive(t) or is_none(t) %}
1125    if not isinstance(fake_dict["fake_key"], {{typename(t)}}):
1126        raise Exception("Not a type of {{typename(t)}}")
1127    {% endif %}
1128    res = {{rvalue(arg(t))}}
1129    ensure(is_bearable(res, {{typename(t)}}), "object is not of type '{{typename(t)}}'")
1130    return res
1131  except Exception as e:
1132    errors.append(f" Failed to deserialize into {{typename(t)}}: {e}")
1133  {% endfor %}
1134  raise SerdeError("Can not deserialize " + repr(data) + " of type " + \
1135          typename(type(data)) + " into {{union_name}}.\\nReasons:\\n" + "\\n".join(errors))
1136""",
1137            "literal": """
1138def {{func}}(cls=cls, maybe_generic=None, maybe_generic_type_vars=None, data=None,
1139             variable_type_args=None, reuse_instances = {{serde_scope.reuse_instances_default}}):
1140  if data in ({%- for v in literal_args -%}{{repr(v)}},{%- endfor -%}):
1141    return data
1142  raise SerdeError("Can not deserialize " + repr(data) + " as {{literal_name}}.")
1143  """,
1144        }
1145    )
1146)
1147
1148
1149def render_from_iter(
1150    cls: type[Any],
1151    legacy_class_deserializer: Optional[DeserializeFunc] = None,
1152    type_check: TypeCheck = strict,
1153    class_deserializer: Optional[ClassDeserializer] = None,
1154) -> str:
1155    renderer = Renderer(
1156        FROM_ITER,
1157        cls=cls,
1158        legacy_class_deserializer=legacy_class_deserializer,
1159        suppress_coerce=(not type_check.is_coerce()),
1160        class_deserializer=class_deserializer,
1161        class_name=typename(cls),
1162    )
1163    fields = list(filter(renderable, defields(cls)))
1164    res = jinja2_env.get_template("iter").render(
1165        func=FROM_ITER,
1166        serde_scope=getattr(cls, SERDE_SCOPE),
1167        fields=fields,
1168        cls_type_vars=get_type_var_names(cls),
1169        rvalue=renderer.render,
1170        arg=to_iter_arg,
1171    )
1172
1173    if renderer.import_numpy:
1174        res = "import numpy\n" + res
1175
1176    return res
1177
1178
1179def get_known_fields(f: DeField[Any], rename_all: Optional[str]) -> list[str]:
1180    names: list[str] = [f.conv_name(rename_all)]
1181    return names + f.alias
1182
1183
1184def render_from_dict(
1185    cls: type[Any],
1186    rename_all: Optional[str] = None,
1187    legacy_class_deserializer: Optional[DeserializeFunc] = None,
1188    type_check: TypeCheck = strict,
1189    class_deserializer: Optional[ClassDeserializer] = None,
1190    deny_unknown_fields: bool = False,
1191) -> str:
1192    renderer = Renderer(
1193        FROM_DICT,
1194        cls=cls,
1195        legacy_class_deserializer=legacy_class_deserializer,
1196        suppress_coerce=(not type_check.is_coerce()),
1197        class_deserializer=class_deserializer,
1198        class_name=typename(cls),
1199    )
1200    fields = list(filter(renderable, defields(cls)))
1201    known_fields = set(
1202        itertools.chain.from_iterable([get_known_fields(f, rename_all) for f in fields])
1203    )
1204    res = jinja2_env.get_template("dict").render(
1205        func=FROM_DICT,
1206        serde_scope=getattr(cls, SERDE_SCOPE),
1207        fields=fields,
1208        type_check=type_check,
1209        cls_type_vars=get_type_var_names(cls),
1210        rvalue=renderer.render,
1211        arg=functools.partial(to_arg, rename_all=rename_all),
1212        deny_unknown_fields=deny_unknown_fields,
1213        known_fields=known_fields,
1214    )
1215
1216    if renderer.import_numpy:
1217        res = "import numpy\n" + res
1218
1219    return res
1220
1221
1222def render_union_func(
1223    cls: type[Any], union_args: Sequence[type[Any]], tagging: Tagging = DefaultTagging
1224) -> str:
1225    union_name = f"Union[{', '.join([typename(a) for a in union_args])}]"
1226
1227    renderer = Renderer(FROM_DICT, cls=cls, suppress_coerce=True)
1228    return jinja2_env.get_template("union").render(
1229        func=union_func_name(UNION_DE_PREFIX, union_args),
1230        serde_scope=getattr(cls, SERDE_SCOPE),
1231        union_args=union_args,
1232        union_name=union_name,
1233        tagging=tagging,
1234        is_taggable=Tagging.is_taggable,
1235        arg=lambda x: DeField(x, datavar="fake_dict", name="fake_key"),
1236        rvalue=renderer.render,
1237        is_primitive=is_primitive,
1238        is_none=is_none,
1239        typename=typename,
1240    )
1241
1242
1243def render_literal_func(
1244    cls: type[Any], literal_args: Sequence[Any], tagging: Tagging = DefaultTagging
1245) -> str:
1246    literal_name = f"Literal[{', '.join([repr(a) for a in literal_args])}]"
1247    return jinja2_env.get_template("literal").render(
1248        func=literal_func_name(literal_args),
1249        serde_scope=getattr(cls, SERDE_SCOPE),
1250        literal_args=literal_args,
1251        literal_name=literal_name,
1252        tagging=tagging,
1253        is_taggable=Tagging.is_taggable,
1254        repr=repr,
1255        type=type,
1256    )