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