serde.se

  1"""
  2This module provides `serialize`, `is_serializable` `to_dict`, `to_tuple` and classes
  3and functions associated with serialization.
  4"""
  5
  6from __future__ import annotations
  7import abc
  8import copy
  9import dataclasses
 10import functools
 11import typing
 12import itertools
 13import jinja2
 14from dataclasses import dataclass, is_dataclass
 15from typing import Any, Generic, Literal, TypeVar
 16from collections.abc import (
 17    Callable,
 18    Iterable,
 19    Iterator,
 20    Mapping,
 21    Set,
 22)
 23
 24from beartype import beartype, BeartypeConf
 25from beartype.door import is_bearable
 26from typing_extensions import dataclass_transform
 27
 28from .compat import (
 29    SerdeError,
 30    SerdeSkip,
 31    T,
 32    get_origin,
 33    is_any,
 34    is_bare_dict,
 35    is_bare_list,
 36    is_bare_opt,
 37    is_bare_set,
 38    is_bare_tuple,
 39    is_class_var,
 40    is_datetime,
 41    is_datetime_instance,
 42    is_dict,
 43    is_enum,
 44    is_generic,
 45    is_list,
 46    is_literal,
 47    is_none,
 48    is_opt,
 49    is_primitive,
 50    is_set,
 51    is_str_serializable,
 52    is_str_serializable_instance,
 53    is_tuple,
 54    is_union,
 55    is_variable_tuple,
 56    is_pep695_type_alias,
 57    iter_types,
 58    iter_unions,
 59    type_args,
 60    typename,
 61)
 62from .core import (
 63    ClassSerializer,
 64    CACHE,
 65    SERDE_SCOPE,
 66    TO_DICT,
 67    TO_ITER,
 68    UNION_SE_PREFIX,
 69    DefaultTagging,
 70    Field,
 71    Scope,
 72    Tagging,
 73    TypeCheck,
 74    add_func,
 75    coerce_object,
 76    disabled,
 77    strict,
 78    fields,
 79    is_instance,
 80    logger,
 81    raise_unsupported_type,
 82    union_func_name,
 83    GLOBAL_CLASS_SERIALIZER,
 84)
 85
 86# Lazy numpy imports to improve startup time
 87
 88__all__ = ["serialize", "is_serializable", "to_dict", "to_tuple"]
 89
 90
 91# Lazy numpy import wrappers to improve startup time
 92def _is_numpy_array(typ: Any) -> bool:
 93    from .numpy import is_numpy_array
 94
 95    return is_numpy_array(typ)
 96
 97
 98def _is_numpy_scalar(typ: Any) -> bool:
 99    from .numpy import is_numpy_scalar
100
101    return is_numpy_scalar(typ)
102
103
104def _is_numpy_jaxtyping(typ: Any) -> bool:
105    from .numpy import is_numpy_jaxtyping
106
107    return is_numpy_jaxtyping(typ)
108
109
110def _is_numpy_datetime(typ: Any) -> bool:
111    from .numpy import is_numpy_datetime
112
113    return is_numpy_datetime(typ)
114
115
116SerializeFunc = Callable[[type[Any], Any], Any]
117""" Interface of Custom serialize function. """
118
119
120def default_serializer(_cls: type[Any], obj: Any) -> Any:
121    """
122    Marker function to tell serde to use the default serializer. It's used when custom serializer
123    is specified at the class but you want to override a field with the default serializer.
124    """
125
126
127def serde_legacy_custom_class_serializer(
128    cls: type[Any], obj: Any, custom: SerializeFunc, default: Callable[[], Any]
129) -> Any:
130    try:
131        return custom(cls, obj)
132    except SerdeSkip:
133        return default()
134
135
136class Serializer(Generic[T], metaclass=abc.ABCMeta):
137    """
138    `Serializer` base class. Subclass this to customize serialize behaviour.
139
140    See `serde.json.JsonSerializer` and `serde.msgpack.MsgPackSerializer` for example usage.
141    """
142
143    @classmethod
144    @abc.abstractmethod
145    def serialize(cls, obj: Any, **opts: Any) -> T:
146        raise NotImplementedError
147
148
149def _make_serialize(
150    cls_name: str,
151    fields: Iterable[str | tuple[str, type[Any]] | tuple[str, type[Any], Any]],
152    *args: Any,
153    rename_all: str | None = None,
154    reuse_instances_default: bool = False,
155    convert_sets_default: bool = False,
156    serializer: SerializeFunc | None = None,
157    tagging: Tagging = DefaultTagging,
158    type_check: TypeCheck = disabled,
159    serialize_class_var: bool = False,
160    class_serializer: ClassSerializer | None = None,
161    **kwargs: Any,
162) -> type[Any]:
163    """
164    Create a serializable class programatically.
165    """
166    C = dataclasses.make_dataclass(cls_name, fields, *args, **kwargs)
167    C = serialize(
168        C,
169        rename_all=rename_all,
170        reuse_instances_default=reuse_instances_default,
171        convert_sets_default=convert_sets_default,
172        serializer=serializer,
173        tagging=tagging,
174        type_check=type_check,
175        serialize_class_var=serialize_class_var,
176        **kwargs,
177    )
178    return C
179
180
181# The `serialize` function can call itself recursively when it needs to generate code for
182# unmarked dataclasses. To avoid infinite recursion, this array remembers types for which code is
183# currently being generated.
184GENERATION_STACK = []
185
186
187@dataclass_transform()
188def serialize(
189    _cls: type[T] | None = None,
190    rename_all: str | None = None,
191    reuse_instances_default: bool = False,
192    convert_sets_default: bool = False,
193    serializer: SerializeFunc | None = None,
194    tagging: Tagging = DefaultTagging,
195    type_check: TypeCheck = strict,
196    serialize_class_var: bool = False,
197    class_serializer: ClassSerializer | None = None,
198    **kwargs: Any,
199) -> type[T]:
200    """
201    A dataclass with this decorator is serializable into any of the data formats
202    supported by pyserde.
203
204    >>> from datetime import datetime
205    >>> from serde import serialize
206    >>> from serde.json import to_json
207    >>>
208    >>> @serialize
209    ... class Foo:
210    ...     i: int
211    ...     s: str
212    ...     f: float
213    ...     b: bool
214    >>>
215    >>> to_json(Foo(i=10, s='foo', f=100.0, b=True))
216    '{"i":10,"s":"foo","f":100.0,"b":true}'
217    """
218
219    def wrap(cls: type[T]) -> type[T]:
220        tagging.check()
221
222        # If no `dataclass` found in the class, dataclassify it automatically.
223        if not is_dataclass(cls):
224            dataclass(cls)
225
226        if type_check.is_strict():
227            serde_beartype = beartype(conf=BeartypeConf(violation_type=SerdeError))
228            serde_beartype(cls)
229
230        g: dict[str, Any] = {}
231
232        # Create a scope storage used by serde.
233        # Each class should get own scope. Child classes can not share scope with parent class.
234        # That's why we need the "scope.cls is not cls" check.
235        scope: Scope | None = getattr(cls, SERDE_SCOPE, None)
236        if scope is None or scope.cls is not cls:
237            scope = Scope(
238                cls,
239                reuse_instances_default=reuse_instances_default,
240                convert_sets_default=convert_sets_default,
241            )
242            setattr(cls, SERDE_SCOPE, scope)
243
244        class_serializers: list[ClassSerializer] = list(
245            itertools.chain(GLOBAL_CLASS_SERIALIZER, [class_serializer] if class_serializer else [])
246        )
247
248        # Set some globals for all generated functions
249        g["cls"] = cls
250        g["copy"] = copy
251        g["serde_scope"] = scope
252        g["SerdeError"] = SerdeError
253        g["raise_unsupported_type"] = raise_unsupported_type
254        g["enum_value"] = enum_value
255        g["is_dataclass"] = is_dataclass
256        g["typename"] = typename  # used in union functions
257        g["is_instance"] = is_instance  # used in union functions
258        g["to_obj"] = to_obj
259        g["typing"] = typing
260        g["Literal"] = Literal
261        g["TypeCheck"] = TypeCheck
262        g["disabled"] = disabled
263        g["coerce_object"] = coerce_object
264        g["class_serializers"] = class_serializers
265        if serializer:
266            g["serde_legacy_custom_class_serializer"] = functools.partial(
267                serde_legacy_custom_class_serializer, custom=serializer
268            )
269
270        # Collect types used in the generated code.
271        for typ in iter_types(cls):
272            # When we encounter a dataclass not marked with serialize, then also generate serialize
273            # functions for it.
274            if is_dataclass_without_se(typ) and typ is not cls:
275                # We call serialize and not wrap to make sure that we will use the default serde
276                # configuration for generating the serialization function.
277                serialize(typ)
278
279            if is_primitive(typ) and not is_enum(typ):
280                continue
281            g[typename(typ)] = typ
282
283        # render all union functions
284        for union in iter_unions(cls):
285            union_args = list(type_args(union))
286            union_key = union_func_name(UNION_SE_PREFIX, union_args)
287            add_func(scope, union_key, render_union_func(cls, union_args, tagging), g)
288            scope.union_se_args[union_key] = union_args
289
290        for f in sefields(cls, serialize_class_var):
291            if f.skip_if:
292                g[f.skip_if.name] = f.skip_if
293            if f.serializer:
294                g[f.serializer.name] = f.serializer
295
296        add_func(
297            scope,
298            TO_ITER,
299            render_to_tuple(cls, serializer, type_check, serialize_class_var, class_serializer),
300            g,
301        )
302        add_func(
303            scope,
304            TO_DICT,
305            render_to_dict(
306                cls, rename_all, serializer, type_check, serialize_class_var, class_serializer
307            ),
308            g,
309        )
310
311        logger.debug(f"{typename(cls)}: {SERDE_SCOPE} {scope}")
312
313        return cls
314
315    if _cls is None:
316        return wrap  # type: ignore
317
318    if _cls in GENERATION_STACK:
319        return _cls
320
321    GENERATION_STACK.append(_cls)
322    try:
323        return wrap(_cls)
324    finally:
325        GENERATION_STACK.pop()
326
327
328def is_serializable(instance_or_class: Any) -> bool:
329    """
330    Test if an instance or class is serializable.
331
332    >>> @serialize
333    ... class Foo:
334    ...     pass
335
336    Testing `Foo` class object returns `True`.
337    >>> is_serializable(Foo)
338    True
339
340    Testing `Foo` object also returns `True`.
341    >>> is_serializable(Foo())
342    True
343    """
344    return hasattr(instance_or_class, SERDE_SCOPE)
345
346
347def is_dataclass_without_se(cls: type[Any]) -> bool:
348    if not dataclasses.is_dataclass(cls):
349        return False
350    if not hasattr(cls, SERDE_SCOPE):
351        return True
352    scope: Scope | None = getattr(cls, SERDE_SCOPE)
353    if not scope:
354        return True
355    return TO_DICT not in scope.funcs
356
357
358def to_obj(
359    o: Any,
360    named: bool,
361    reuse_instances: bool | None = None,
362    convert_sets: bool | None = None,
363    skip_none: bool = False,
364    c: Any | None = None,
365) -> Any:
366    def serializable_to_obj(object: Any) -> Any:
367        serde_scope: Scope = getattr(object, SERDE_SCOPE)
368        func_name = TO_DICT if named else TO_ITER
369        return serde_scope.funcs[func_name](
370            object,
371            reuse_instances=reuse_instances,
372            convert_sets=convert_sets,
373            skip_none=skip_none,
374        )
375
376    try:
377        thisfunc = functools.partial(
378            to_obj,
379            named=named,
380            reuse_instances=reuse_instances,
381            convert_sets=convert_sets,
382            skip_none=skip_none,
383        )
384
385        # If a class in the argument is a non-dataclass class e.g. Union[Foo, Bar],
386        # pyserde generates a wrapper (de)serializable dataclass on the fly,
387        # and use it to serialize the object.
388        if c and is_union(c) and not is_opt(c):
389            return CACHE.serialize_union(c, o)
390
391        if o is None:
392            return None
393        if is_dataclass_without_se(o):
394            # Do not automatically implement beartype if dataclass without serde decorator
395            # is passed, because it is surprising for users
396            # See https://github.com/yukinarit/pyserde/issues/480
397            serialize(type(o), type_check=disabled)
398            return serializable_to_obj(o)
399        elif is_serializable(o):
400            return serializable_to_obj(o)
401        elif is_bearable(o, list):  # type: ignore[arg-type]  # pyright: ignore[reportArgumentType]
402            return [thisfunc(e) for e in o]
403        elif is_bearable(o, tuple):  # type: ignore[arg-type]  # pyright: ignore[reportArgumentType]
404            return tuple(thisfunc(e) for e in o)
405        elif isinstance(o, Mapping):
406            return {k: thisfunc(v) for k, v in o.items()}
407        elif isinstance(o, Set):
408            return [thisfunc(e) for e in o]
409        elif is_str_serializable_instance(o) or is_datetime_instance(o):
410            se_cls = o.__class__ if not c or c is Any else c
411            return CACHE.serialize(
412                se_cls,
413                o,
414                reuse_instances=reuse_instances,
415                convert_sets=convert_sets,
416                skip_none=skip_none,
417            )
418
419        return o
420
421    except Exception as e:
422        raise SerdeError(e) from None
423
424
425def astuple(v: Any) -> tuple[Any, ...]:
426    """
427    Serialize object into tuple.
428    """
429    return to_tuple(v, reuse_instances=False, convert_sets=False)
430
431
432def to_tuple(
433    o: Any,
434    c: type[Any] | None = None,
435    reuse_instances: bool | None = None,
436    convert_sets: bool | None = None,
437    skip_none: bool = False,
438) -> tuple[Any, ...]:
439    """
440    Serialize object into tuple.
441
442    >>> @serialize
443    ... class Foo:
444    ...     i: int
445    ...     s: str = 'foo'
446    ...     f: float = 100.0
447    ...     b: bool = True
448    >>>
449    >>> to_tuple(Foo(i=10))
450    (10, 'foo', 100.0, True)
451
452    You can pass any type supported by pyserde. For example,
453
454    >>> lst = [Foo(i=10), Foo(i=20)]
455    >>> to_tuple(lst)
456    [(10, 'foo', 100.0, True), (20, 'foo', 100.0, True)]
457    """
458    return to_obj(  # type: ignore
459        o,
460        named=False,
461        c=c,
462        reuse_instances=reuse_instances,
463        convert_sets=convert_sets,
464        skip_none=skip_none,
465    )
466
467
468def asdict(v: Any) -> dict[Any, Any]:
469    """
470    Serialize object into dictionary.
471    """
472    return to_dict(v, reuse_instances=False, convert_sets=False)
473
474
475def to_dict(
476    o: Any,
477    c: type[Any] | None = None,
478    reuse_instances: bool | None = None,
479    convert_sets: bool | None = None,
480    skip_none: bool = False,
481) -> dict[Any, Any]:
482    """
483    Serialize object into python dictionary. This function ensures that the dataclass's fields are
484    accurately represented as key-value pairs in the resulting dictionary.
485
486    * `o`: Any pyserde object that you want to convert to `dict`
487    * `c`: Optional class argument
488    * `reuse_instances`: pyserde will pass instances (e.g. Path, datetime) directly to serializer
489    instead of converting them to serializable representation e.g. string. This behaviour allows
490    to delegate serializtation to underlying data format packages e.g. `pyyaml` and potentially
491    improve performance.
492    * `convert_sets`: This option controls how sets are handled during serialization and
493    deserialization. When `convert_sets` is set to True, pyserde will convert sets to lists during
494    serialization and back to sets during deserialization. This is useful for data formats that
495    do not natively support sets.
496    * `skip_none`: When set to True, any field in the class with a None value is excluded from the
497    serialized output. Defaults to False.
498
499    >>> from serde import serde
500    >>> @serde
501    ... class Foo:
502    ...     i: int
503    ...     s: str = 'foo'
504    ...     f: float = 100.0
505    ...     b: bool = True
506    >>>
507    >>> to_dict(Foo(i=10))
508    {'i': 10, 's': 'foo', 'f': 100.0, 'b': True}
509
510    You can serialize not only pyserde objects but also objects of any supported types. For example,
511    the following example serializes list of pyserde objects into dict.
512
513    >>> lst = [Foo(i=10), Foo(i=20)]
514    >>> to_dict(lst)
515    [{'i': 10, 's': 'foo', 'f': 100.0, 'b': True}, {'i': 20, 's': 'foo', 'f': 100.0, 'b': True}]
516    """
517    return to_obj(  # type: ignore
518        o,
519        named=True,
520        c=c,
521        reuse_instances=reuse_instances,
522        convert_sets=convert_sets,
523        skip_none=skip_none,
524    )
525
526
527@dataclass
528class SeField(Field[T]):
529    """
530    Field class for serialization.
531    """
532
533    @property
534    def varname(self) -> str:
535        """
536        Get variable name in the generated code e.g. obj.a.b
537        """
538        var = getattr(self.parent, "varname", None) if self.parent else None
539        if var:
540            return f"{var}.{self.name}"
541        else:
542            if self.name is None:
543                raise SerdeError("Field name is None.")
544            return self.name
545
546    def __getitem__(self, n: int) -> SeField[Any]:
547        typ = type_args(self.type)[n]
548        opts: dict[str, Any] = {
549            "kw_only": self.kw_only,
550            "case": self.case,
551            "alias": self.alias,
552            "rename": self.rename,
553            "skip": self.skip,
554            "skip_if": self.skip_if,
555            "skip_if_false": self.skip_if_false,
556            "skip_if_default": self.skip_if_default,
557            "serializer": self.serializer,
558            "deserializer": self.deserializer,
559            "flatten": self.flatten,
560        }
561        return SeField(typ, name=None, **opts)
562
563
564def sefields(cls: type[Any], serialize_class_var: bool = False) -> Iterator[SeField[Any]]:
565    """
566    Iterate fields for serialization.
567    """
568    for f in fields(SeField, cls, serialize_class_var=serialize_class_var):
569        f.parent = SeField(None, "obj")  # type: ignore
570        yield f
571
572
573jinja2_env = jinja2.Environment(
574    loader=jinja2.DictLoader(
575        {
576            "dict": """
577def {{func}}(obj, reuse_instances = None, convert_sets = None, skip_none = False):
578  if reuse_instances is None:
579    reuse_instances = {{serde_scope.reuse_instances_default}}
580  if convert_sets is None:
581    convert_sets = {{serde_scope.convert_sets_default}}
582  if not is_dataclass(obj):
583    return copy.deepcopy(obj)
584
585  res = {}
586  {% for f in fields -%}
587  subres = {{rvalue(f)}}
588  {% if not f.skip -%}
589    {% if f.skip_if -%}
590  if not {{f.skip_if.name}}(subres):
591    {{lvalue(f)}} = subres
592    {% else -%}
593  if skip_none:
594    if subres is not None:
595      {{lvalue(f)}} = subres
596  else:
597    {{lvalue(f)}} = subres
598    {% endif -%}
599  {% endif %}
600
601  {% endfor -%}
602  return res
603""",
604            "iter": """
605def {{func}}(obj, reuse_instances=None, convert_sets=None, skip_none=False):
606  if reuse_instances is None:
607    reuse_instances = {{serde_scope.reuse_instances_default}}
608  if convert_sets is None:
609    convert_sets = {{serde_scope.convert_sets_default}}
610  if not is_dataclass(obj):
611    return copy.deepcopy(obj)
612
613  return (
614  {% for f in fields -%}
615  {% if not f.skip|default(False) %}
616  {{rvalue(f)}},
617  {% endif -%}
618  {% endfor -%}
619  )
620""",
621            "union": """
622def {{func}}(obj, reuse_instances, convert_sets, skip_none=False):
623  union_args = serde_scope.union_se_args['{{func}}']
624
625  {% for t in union_args %}
626  if is_instance(obj, union_args[{{loop.index0}}]):
627    {% if tagging.is_external() and is_taggable(t) %}
628    return {"{{typename(t)}}": {{rvalue(arg(t))}}}
629
630    {% elif tagging.is_internal() and is_taggable(t) %}
631    res = {{rvalue(arg(t))}}
632    res["{{tagging.tag}}"] = "{{typename(t)}}"
633    return res
634
635    {% elif tagging.is_adjacent() and is_taggable(t) %}
636    res = {"{{tagging.content}}": {{rvalue(arg(t))}}}
637    res["{{tagging.tag}}"] = "{{typename(t)}}"
638    return res
639
640    {% else %}
641    return {{rvalue(arg(t))}}
642    {% endif %}
643  {% endfor %}
644  raise SerdeError("Can not serialize " + \
645                   repr(obj) + \
646                   " of type " + \
647                   typename(type(obj)) + \
648                   " for {{union_name}}")
649""",
650        }
651    )
652)
653
654
655def render_to_tuple(
656    cls: type[Any],
657    legacy_class_serializer: SerializeFunc | None = None,
658    type_check: TypeCheck = strict,
659    serialize_class_var: bool = False,
660    class_serializer: ClassSerializer | None = None,
661) -> str:
662    renderer = Renderer(
663        TO_ITER,
664        legacy_class_serializer,
665        suppress_coerce=(not type_check.is_coerce()),
666        serialize_class_var=serialize_class_var,
667        class_serializer=class_serializer,
668        class_name=typename(cls),
669    )
670    return jinja2_env.get_template("iter").render(
671        func=TO_ITER,
672        serde_scope=getattr(cls, SERDE_SCOPE),
673        fields=sefields(cls, serialize_class_var),
674        type_check=type_check,
675        rvalue=renderer.render,
676    )
677
678
679def render_to_dict(
680    cls: type[Any],
681    case: str | None = None,
682    legacy_class_serializer: SerializeFunc | None = None,
683    type_check: TypeCheck = strict,
684    serialize_class_var: bool = False,
685    class_serializer: ClassSerializer | None = None,
686) -> str:
687    renderer = Renderer(
688        TO_DICT,
689        legacy_class_serializer,
690        suppress_coerce=(not type_check.is_coerce()),
691        class_serializer=class_serializer,
692        class_name=typename(cls),
693    )
694    lrenderer = LRenderer(case, serialize_class_var)
695    return jinja2_env.get_template("dict").render(
696        func=TO_DICT,
697        serde_scope=getattr(cls, SERDE_SCOPE),
698        fields=sefields(cls, serialize_class_var),
699        type_check=type_check,
700        lvalue=lrenderer.render,
701        rvalue=renderer.render,
702    )
703
704
705def render_union_func(
706    cls: type[Any], union_args: list[type[Any]], tagging: Tagging = DefaultTagging
707) -> str:
708    """
709    Render function that serializes a field with union type.
710    """
711    union_name = f"Union[{', '.join([typename(a) for a in union_args])}]"
712    renderer = Renderer(TO_DICT, suppress_coerce=True, class_name=typename(cls))
713    return jinja2_env.get_template("union").render(
714        func=union_func_name(UNION_SE_PREFIX, union_args),
715        serde_scope=getattr(cls, SERDE_SCOPE),
716        union_args=union_args,
717        union_name=union_name,
718        tagging=tagging,
719        is_taggable=Tagging.is_taggable,
720        arg=lambda x: SeField(x, "obj"),
721        rvalue=renderer.render,
722        typename=typename,
723    )
724
725
726@dataclass
727class LRenderer:
728    """
729    Render lvalue for various types.
730    """
731
732    case: str | None
733    serialize_class_var: bool = False
734
735    def render(self, arg: SeField[Any]) -> str:
736        """
737        Render lvalue
738        """
739        if is_dataclass(arg.type) and arg.flatten:
740            return self.flatten(arg)
741        elif is_opt(arg.type) and arg.flatten:
742            inner = arg[0]
743            if is_dataclass(inner.type):
744                return self.flatten(inner)
745
746        return f'res["{arg.conv_name(self.case)}"]'
747
748    def flatten(self, arg: SeField[Any]) -> str:
749        """
750        Render field with flatten attribute.
751        """
752        flattened = []
753        for f in sefields(arg.type, self.serialize_class_var):
754            flattened.append(self.render(f))
755        return ", ".join(flattened)
756
757
758@dataclass
759class Renderer:
760    """
761    Render rvalue for code generation.
762    """
763
764    func: str
765    legacy_class_serializer: SerializeFunc | None = None
766    suppress_coerce: bool = False
767    """ Suppress type coercing because generated union serializer has its own type checking """
768    serialize_class_var: bool = False
769    class_serializer: ClassSerializer | None = None
770    class_name: str | None = None
771
772    def render(self, arg: SeField[Any]) -> str:
773        """
774        Render rvalue
775        """
776        implemented_methods: dict[type[Any], int] = {}
777        class_serializers: Iterable[ClassSerializer] = itertools.chain(
778            GLOBAL_CLASS_SERIALIZER, [self.class_serializer] if self.class_serializer else []
779        )
780        for n, class_serializer in enumerate(class_serializers):
781            for method in class_serializer.__class__.serialize.methods:  # type: ignore
782                implemented_methods[method.signature.types[1]] = n
783
784        custom_serializer_available = arg.type in implemented_methods
785        if custom_serializer_available and not arg.serializer:
786            res = f"class_serializers[{implemented_methods[arg.type]}].serialize({arg.varname})"
787        elif arg.serializer and arg.serializer.inner is not default_serializer:
788            res = self.custom_field_serializer(arg)
789        elif is_dataclass(arg.type):
790            res = self.dataclass(arg)
791        elif is_opt(arg.type):
792            res = self.opt(arg)
793        elif is_list(arg.type):
794            res = self.list(arg)
795        elif is_set(arg.type):
796            res = self.set(arg)
797        elif is_dict(arg.type):
798            res = self.dict(arg)
799        elif is_tuple(arg.type):
800            res = self.tuple(arg)
801        elif is_enum(arg.type):
802            res = self.enum(arg)
803        elif _is_numpy_datetime(arg.type):
804            from .numpy import serialize_numpy_datetime
805
806            res = serialize_numpy_datetime(arg)
807        elif _is_numpy_scalar(arg.type):
808            from .numpy import serialize_numpy_scalar
809
810            res = serialize_numpy_scalar(arg)
811        elif _is_numpy_array(arg.type):
812            from .numpy import serialize_numpy_array
813
814            res = serialize_numpy_array(arg)
815        elif _is_numpy_jaxtyping(arg.type):
816            from .numpy import serialize_numpy_array
817
818            res = serialize_numpy_array(arg)
819        elif is_primitive(arg.type):
820            res = self.primitive(arg)
821        elif is_union(arg.type):
822            res = self.union_func(arg)
823        elif is_str_serializable(arg.type):
824            res = f"{arg.varname} if reuse_instances else {self.string(arg)}"
825        elif is_datetime(arg.type):
826            res = f"{arg.varname} if reuse_instances else {arg.varname}.isoformat()"
827        elif is_none(arg.type):
828            res = "None"
829        elif is_any(arg.type) or is_bearable(arg.type, TypeVar):  # type: ignore[arg-type]  # pyright: ignore
830            res = f"to_obj({arg.varname}, True, False, False, skip_none, typing.Any)"
831        elif is_generic(arg.type):
832            origin = get_origin(arg.type)
833            assert origin
834            arg.type = origin
835            res = self.render(arg)
836        elif is_literal(arg.type):
837            res = self.literal(arg)
838        elif is_class_var(arg.type):
839            arg.type = type_args(arg.type)[0]
840            res = self.render(arg)
841        elif is_pep695_type_alias(arg.type):
842            res = self.render(dataclasses.replace(arg, type=arg.type.__value__))
843        else:
844            res = f"raise_unsupported_type({arg.varname})"
845
846        # Custom field serializer overrides custom class serializer.
847        if self.legacy_class_serializer and not arg.serializer and not custom_serializer_available:
848            return (
849                "serde_legacy_custom_class_serializer("
850                f"{typename(arg.type)}, "
851                f"{arg.varname}, "
852                f"default=lambda: {res})"
853            )
854        else:
855            return res
856
857    def custom_field_serializer(self, arg: SeField[Any]) -> str:
858        """
859        Render rvalue for the field with custom serializer.
860        """
861        assert arg.serializer
862        return f"{arg.serializer.name}({arg.varname})"
863
864    def dataclass(self, arg: SeField[Any]) -> str:
865        """
866        Render rvalue for dataclass.
867        """
868        if arg.flatten:
869            flattened = []
870            for f in sefields(arg.type, self.serialize_class_var):
871                f.parent = arg
872                flattened.append(self.render(f))
873            return ", ".join(flattened)
874        else:
875            return (
876                f"{arg.varname}.{SERDE_SCOPE}.funcs['{self.func}']({arg.varname}, "
877                "reuse_instances=reuse_instances, convert_sets=convert_sets, skip_none=skip_none)"
878            )
879
880    def opt(self, arg: SeField[Any]) -> str:
881        """
882        Render rvalue for optional.
883        """
884        if is_bare_opt(arg.type):
885            return f"{arg.varname} if {arg.varname} is not None else None"
886
887        inner = arg[0]
888        inner.name = arg.varname
889        if arg.flatten:
890            return self.render(inner)
891        else:
892            return f"({self.render(inner)}) if {arg.varname} is not None else None"
893
894    def list(self, arg: SeField[Any]) -> str:
895        """
896        Render rvalue for list.
897        """
898        if is_bare_list(arg.type):
899            origin = get_origin(arg.type) or arg.type
900            return arg.varname if origin is list else f"list({arg.varname})"
901        else:
902            earg = arg[0]
903            earg.name = "v"
904            return f"[{self.render(earg)} for v in {arg.varname}]"
905
906    def set(self, arg: SeField[Any]) -> str:
907        """
908        Render rvalue for set.
909        """
910        if is_bare_set(arg.type):
911            return f"list({arg.varname}) if convert_sets else {arg.varname}"
912        else:
913            earg = arg[0]
914            earg.name = "v"
915            return (
916                f"[{self.render(earg)} for v in {arg.varname}] "
917                f"if convert_sets else set({self.render(earg)} for v in {arg.varname})"
918            )
919
920    def tuple(self, arg: SeField[Any]) -> str:
921        """
922        Render rvalue for tuple.
923        """
924        if is_bare_tuple(arg.type):
925            return arg.varname
926        elif is_variable_tuple(arg.type):
927            earg = arg[0]
928            earg.name = "v"
929            return f"tuple({self.render(earg)} for v in {arg.varname})"
930        else:
931            rvalues = []
932            for i, _ in enumerate(type_args(arg.type)):
933                r = arg[i]
934                r.name = f"{arg.varname}[{i}]"
935                rvalues.append(self.render(r))
936            return f"({', '.join(rvalues)},)"  # trailing , is required for single element tuples
937
938    def dict(self, arg: SeField[Any]) -> str:
939        """
940        Render rvalue for dict.
941        """
942        if is_bare_dict(arg.type):
943            return arg.varname
944        else:
945            karg = arg[0]
946            karg.name = "k"
947            varg = arg[1]
948            varg.name = "v"
949            return f"{{{self.render(karg)}: {self.render(varg)} for k, v in {arg.varname}.items()}}"
950
951    def enum(self, arg: SeField[Any]) -> str:
952        return f"enum_value({typename(arg.type)}, {arg.varname})"
953
954    def primitive(self, arg: SeField[Any]) -> str:
955        """
956        Render rvalue for primitives.
957        """
958        typ = typename(arg.type)
959        var = arg.varname
960        if self.suppress_coerce:
961            return var
962        else:
963            assert arg.name
964            escaped_arg_name = arg.name.replace('"', '\\"')
965            return f'coerce_object("{self.class_name}", "{escaped_arg_name}", {typ}, {var})'
966
967    def string(self, arg: SeField[Any]) -> str:
968        return f"str({arg.varname})"
969
970    def union_func(self, arg: SeField[Any]) -> str:
971        func_name = union_func_name(UNION_SE_PREFIX, list(type_args(arg.type)))
972        return (
973            f"serde_scope.funcs['{func_name}']({arg.varname}, "
974            "reuse_instances, convert_sets, skip_none)"
975        )
976
977    def literal(self, arg: SeField[Any]) -> str:
978        return f"{arg.varname}"
979
980
981def enum_value(cls: Any, e: Any) -> Any:
982    """
983    Helper function to get value from enum or enum compatible value.
984    """
985    if is_enum(e):
986        v = e.value
987        # Recursively get value of Nested enum.
988        if is_enum(v):
989            return enum_value(v.__class__, v)
990        else:
991            return v
992    else:
993        return cls(e).value