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