serde.core

   1"""
   2pyserde core module.
   3"""
   4
   5from __future__ import annotations
   6import dataclasses
   7import enum
   8import functools
   9import logging
  10import re
  11import casefy
  12from dataclasses import dataclass
  13
  14from beartype.door import is_bearable
  15from collections import deque, Counter
  16from collections.abc import Mapping, Sequence, MutableSequence, Set, Callable, Hashable
  17from typing import (
  18    overload,
  19    TypeVar,
  20    Generic,
  21    Any,
  22    Protocol,
  23    get_type_hints,
  24    cast,
  25)
  26
  27from .compat import (
  28    T,
  29    SerdeError,
  30    dataclass_fields,
  31    get_origin,
  32    is_bare_counter,
  33    is_bare_deque,
  34    is_bare_dict,
  35    is_bare_list,
  36    is_bare_set,
  37    is_bare_tuple,
  38    is_class_var,
  39    is_counter,
  40    is_deque,
  41    is_dict,
  42    is_flatten_dict,
  43    is_generic,
  44    is_list,
  45    is_literal,
  46    is_new_type_primitive,
  47    is_any,
  48    is_opt,
  49    is_opt_dataclass,
  50    is_set,
  51    is_tuple,
  52    is_union,
  53    is_variable_tuple,
  54    type_args,
  55    typename,
  56    _WithTagging,
  57)
  58
  59__all__ = [
  60    "Scope",
  61    "gen",
  62    "add_func",
  63    "Func",
  64    "Field",
  65    "fields",
  66    "FlattenOpts",
  67    "conv",
  68    "union_func_name",
  69]
  70
  71logger = logging.getLogger("serde")
  72
  73
  74# name of the serde context key
  75SERDE_SCOPE = "__serde__"
  76
  77# main function keys
  78FROM_ITER = "from_iter"
  79FROM_DICT = "from_dict"
  80TO_ITER = "to_iter"
  81TO_DICT = "to_dict"
  82TYPE_CHECK = "typecheck"
  83
  84# prefixes used to distinguish the direction of a union function
  85UNION_SE_PREFIX = "union_se"
  86UNION_DE_PREFIX = "union_de"
  87
  88LITERAL_DE_PREFIX = "literal_de"
  89
  90SETTINGS = {"debug": False}
  91
  92
  93@dataclass(frozen=True)
  94class UnionCacheKey:
  95    union: Hashable
  96    tagging: Tagging
  97
  98
  99def init(debug: bool = False) -> None:
 100    SETTINGS["debug"] = debug
 101
 102
 103@dataclass
 104class Cache:
 105    """
 106    Cache the generated code for non-dataclass classes.
 107
 108    for example, a type not bound in a dataclass is passed in from_json
 109
 110    ```
 111    from_json(Union[Foo, Bar], ...)
 112    ```
 113
 114    It creates the following wrapper dataclass on the fly,
 115
 116    ```
 117    @serde
 118    @dataclass
 119    class Union_Foo_bar:
 120        v: Union[Foo, Bar]
 121    ```
 122
 123    Then store this class in this cache. Whenever the same type is passed,
 124    the class is retrieved from this cache. So the overhead of the codegen
 125    should be only once.
 126    """
 127
 128    classes: dict[Hashable, type[Any]] = dataclasses.field(default_factory=dict)
 129
 130    def _get_class(self, cls: type[Any]) -> type[Any]:
 131        """
 132        Get a wrapper class from the the cache. If not found, it will generate
 133        the class and store it in the cache.
 134        """
 135        wrapper = self.classes.get(cls)  # type: ignore[call-overload] # mypy doesn't recognize type[Any] as Hashable
 136        return wrapper or self._generate_class(cls)
 137
 138    def _generate_class(self, cls: type[Any]) -> type[Any]:
 139        """
 140        Generate a wrapper dataclass then make the it (de)serializable using
 141        @serde decorator.
 142        """
 143        from . import serde
 144
 145        class_name = f"Wrapper{typename(cls)}"
 146        logger.debug(f"Generating a wrapper class code for {class_name}")
 147
 148        wrapper = dataclasses.make_dataclass(class_name, [("v", cls)])
 149
 150        serde(wrapper)
 151        self.classes[cls] = wrapper  # type: ignore[index] # mypy doesn't recognize type[Any] as Hashable
 152
 153        logger.debug(f"(de)serializing code for {class_name} was generated")
 154        return wrapper
 155
 156    def serialize(self, cls: type[Any], obj: Any, **kwargs: Any) -> Any:
 157        """
 158        Serialize the specified type of object into dict or tuple.
 159        """
 160        wrapper = self._get_class(cls)
 161        scope: Scope = getattr(wrapper, SERDE_SCOPE)
 162        data = scope.funcs[TO_DICT](wrapper(obj), **kwargs)
 163
 164        logging.debug(f"Intermediate value: {data}")
 165
 166        return data["v"]
 167
 168    def deserialize(self, cls: type[T], obj: Any) -> T:
 169        """
 170        Deserialize from dict or tuple into the specified type.
 171        """
 172        wrapper = self._get_class(cls)
 173        scope: Scope = getattr(wrapper, SERDE_SCOPE)
 174        return scope.funcs[FROM_DICT](data={"v": obj}).v  # type: ignore
 175
 176    def _get_union_class(self, cls: type[Any]) -> type[Any] | None:
 177        """
 178        Get a wrapper class from the the cache. If not found, it will generate
 179        the class and store it in the cache.
 180        """
 181        union_cls, tagging = _extract_from_with_tagging(cls)
 182        cache_key = UnionCacheKey(union=union_cls, tagging=tagging)
 183        wrapper = self.classes.get(cache_key)
 184        return wrapper or self._generate_union_class(cls)
 185
 186    def _generate_union_class(self, cls: type[Any]) -> type[Any]:
 187        """
 188        Generate a wrapper dataclass then make the it (de)serializable using
 189        @serde decorator.
 190        """
 191        import serde
 192
 193        union_cls, tagging = _extract_from_with_tagging(cls)
 194        cache_key = UnionCacheKey(union=union_cls, tagging=tagging)
 195        class_name = union_func_name(
 196            f"{tagging.produce_unique_class_name()}Union", list(type_args(union_cls))
 197        )
 198        wrapper = dataclasses.make_dataclass(class_name, [("v", union_cls)])
 199        serde.serde(wrapper, tagging=tagging)
 200        self.classes[cache_key] = wrapper
 201        return wrapper
 202
 203    def serialize_union(self, cls: type[Any], obj: Any) -> Any:
 204        """
 205        Serialize the specified Union into dict or tuple.
 206        """
 207        union_cls, _ = _extract_from_with_tagging(cls)
 208        wrapper = self._get_union_class(cls)
 209        scope: Scope = getattr(wrapper, SERDE_SCOPE)
 210        func_name = union_func_name(UNION_SE_PREFIX, list(type_args(union_cls)))
 211        return scope.funcs[func_name](obj, False, False)
 212
 213    def deserialize_union(
 214        self,
 215        cls: type[T],
 216        data: Any,
 217        deserialize_numbers: Callable[[str | int], float] | None = None,
 218    ) -> T:
 219        """
 220        Deserialize from dict or tuple into the specified Union.
 221        """
 222        union_cls, _ = _extract_from_with_tagging(cls)
 223        wrapper = self._get_union_class(cls)
 224        scope: Scope = getattr(wrapper, SERDE_SCOPE)
 225        func_name = union_func_name(UNION_DE_PREFIX, list(type_args(union_cls)))
 226        return cast(
 227            T,
 228            scope.funcs[func_name](
 229                cls=union_cls, data=data, deserialize_numbers=deserialize_numbers
 230            ),
 231        )
 232
 233
 234def _extract_from_with_tagging(maybe_with_tagging: Any) -> tuple[Any, Tagging]:
 235    if isinstance(maybe_with_tagging, _WithTagging):
 236        return maybe_with_tagging.inner, maybe_with_tagging.tagging
 237    else:
 238        return maybe_with_tagging, ExternalTagging
 239
 240
 241CACHE = Cache()
 242""" Global cache variable for non-dataclass classes """
 243
 244
 245@dataclass
 246class Scope:
 247    """
 248    Container to store types and functions used in code generation context.
 249    """
 250
 251    cls: type[Any]
 252    """ The exact class this scope is for
 253    (needed to distinguish scopes between inherited classes) """
 254
 255    funcs: dict[str, Callable[..., Any]] = dataclasses.field(default_factory=dict)
 256    """ Generated serialize and deserialize functions """
 257
 258    defaults: dict[str, Callable[..., Any] | Any] = dataclasses.field(default_factory=dict)
 259    """ Default values of the dataclass fields (factories & normal values) """
 260
 261    code: dict[str, str] = dataclasses.field(default_factory=dict)
 262    """ Generated source code (only filled when debug is True) """
 263
 264    union_se_args: dict[str, list[type[Any]]] = dataclasses.field(default_factory=dict)
 265    """ The union serializing functions need references to their types """
 266
 267    reuse_instances_default: bool = True
 268    """ Default values for to_dict & from_dict arguments """
 269
 270    convert_sets_default: bool = False
 271
 272    transparent: bool = False
 273    """If True, serialize/deserialize as the single inner field (serde-rs `transparent`)."""
 274
 275    def __repr__(self) -> str:
 276        res: list[str] = []
 277
 278        res.append("==================================================")
 279        res.append(self._justify(self.cls.__name__))
 280        res.append("==================================================")
 281        res.append("")
 282
 283        if self.code:
 284            res.append("--------------------------------------------------")
 285            res.append(self._justify("Functions generated by pyserde"))
 286            res.append("--------------------------------------------------")
 287            res.extend(list(self.code.values()))
 288            res.append("")
 289
 290        if self.funcs:
 291            res.append("--------------------------------------------------")
 292            res.append(self._justify("Function references in scope"))
 293            res.append("--------------------------------------------------")
 294            for k, v in self.funcs.items():
 295                res.append(f"{k}: {v}")
 296            res.append("")
 297
 298        if self.defaults:
 299            res.append("--------------------------------------------------")
 300            res.append(self._justify("Default values for the dataclass fields"))
 301            res.append("--------------------------------------------------")
 302            for k, v in self.defaults.items():
 303                res.append(f"{k}: {v}")
 304            res.append("")
 305
 306        if self.union_se_args:
 307            res.append("--------------------------------------------------")
 308            res.append(self._justify("Type list by used for union serialize functions"))
 309            res.append("--------------------------------------------------")
 310            for k, lst in self.union_se_args.items():
 311                res.append(f"{k}: {list(lst)}")
 312            res.append("")
 313
 314        return "\n".join(res)
 315
 316    def _justify(self, s: str, length: int = 50) -> str:
 317        white_spaces = int((50 - len(s)) / 2)
 318        return " " * (white_spaces if white_spaces > 0 else 0) + s
 319
 320
 321def raise_unsupported_type(obj: Any) -> None:
 322    # needed because we can not render a raise statement everywhere, e.g. as argument
 323    raise SerdeError(f"Unsupported type: {typename(type(obj))}")
 324
 325
 326def gen(
 327    code: str, globals: dict[str, Any] | None = None, locals: dict[str, Any] | None = None
 328) -> str:
 329    """
 330    A wrapper of builtin `exec` function.
 331    """
 332    if SETTINGS["debug"]:
 333        # black formatting is only important when debugging
 334        try:
 335            from black import FileMode, format_str
 336
 337            code = format_str(code, mode=FileMode(line_length=100))
 338        except Exception:
 339            pass
 340    exec(code, globals, locals)
 341    return code
 342
 343
 344def add_func(serde_scope: Scope, func_name: str, func_code: str, globals: dict[str, Any]) -> None:
 345    """
 346    Generate a function and add it to a Scope's `funcs` dictionary.
 347
 348    * `serde_scope`: the Scope instance to modify
 349    * `func_name`: the name of the function
 350    * `func_code`: the source code of the function
 351    * `globals`: global variables that should be accessible to the generated function
 352    """
 353
 354    code = gen(func_code, globals)
 355    serde_scope.funcs[func_name] = globals[func_name]
 356
 357    if SETTINGS["debug"]:
 358        serde_scope.code[func_name] = code
 359
 360
 361def is_instance(obj: Any, typ: Any) -> bool:
 362    """
 363    pyserde's own `isinstance` helper. It accepts subscripted generics e.g. `list[int]` and
 364    deeply check object against declared type.
 365    """
 366    if dataclasses.is_dataclass(typ):
 367        if not isinstance(typ, type):
 368            raise SerdeError("expect dataclass class but dataclass instance received")
 369        return isinstance(obj, typ)
 370    elif is_opt(typ):
 371        return is_opt_instance(obj, typ)
 372    elif is_union(typ):
 373        return is_union_instance(obj, typ)
 374    elif is_list(typ):
 375        return is_list_instance(obj, typ)
 376    elif is_set(typ):
 377        return is_set_instance(obj, typ)
 378    elif is_tuple(typ):
 379        return is_tuple_instance(obj, typ)
 380    elif is_counter(typ):
 381        # Counter must be checked before dict since Counter is a subclass of dict
 382        return is_counter_instance(obj, typ)
 383    elif is_dict(typ):
 384        return is_dict_instance(obj, typ)
 385    elif is_deque(typ):
 386        return is_deque_instance(obj, typ)
 387    elif is_generic(typ):
 388        return is_generic_instance(obj, typ)
 389    elif is_literal(typ):
 390        return True
 391    elif is_new_type_primitive(typ):
 392        inner = getattr(typ, "__supertype__", None)
 393        if type(inner) is type:
 394            return isinstance(obj, inner)
 395        else:
 396            return False
 397    elif is_any(typ):
 398        return True
 399    elif typ is Ellipsis:
 400        return True
 401    else:
 402        return is_bearable(obj, typ)
 403
 404
 405def is_opt_instance(obj: Any, typ: type[Any]) -> bool:
 406    if obj is None:
 407        return True
 408    opt_arg = type_args(typ)[0]
 409    return is_instance(obj, opt_arg)
 410
 411
 412def is_union_instance(obj: Any, typ: type[Any]) -> bool:
 413    for arg in type_args(typ):
 414        if is_instance(obj, arg):
 415            return True
 416    return False
 417
 418
 419def is_list_instance(obj: Any, typ: type[Any]) -> bool:
 420    origin = get_origin(typ) or typ
 421    if origin is list:
 422        if not isinstance(obj, list):
 423            return False
 424    elif origin is MutableSequence:
 425        if isinstance(obj, (str, bytes, bytearray)):
 426            return False
 427        if not isinstance(obj, MutableSequence):
 428            return False
 429    else:
 430        if isinstance(obj, (str, bytes, bytearray)):
 431            return False
 432        if not isinstance(obj, Sequence):
 433            return False
 434    if len(obj) == 0 or is_bare_list(typ):
 435        return True
 436    list_arg = type_args(typ)[0]
 437    # for speed reasons we just check the type of the 1st element
 438    return is_instance(obj[0], list_arg)
 439
 440
 441def is_set_instance(obj: Any, typ: type[Any]) -> bool:
 442    if not isinstance(obj, Set):
 443        return False
 444    if len(obj) == 0 or is_bare_set(typ):
 445        return True
 446    set_arg = type_args(typ)[0]
 447    # for speed reasons we just check the type of the 1st element
 448    return is_instance(next(iter(obj)), set_arg)
 449
 450
 451def is_tuple_instance(obj: Any, typ: type[Any]) -> bool:
 452    args = type_args(typ)
 453
 454    if not isinstance(obj, tuple):
 455        return False
 456
 457    # empty tuple
 458    if len(args) == 0 and len(obj) == 0:
 459        return True
 460
 461    # In the form of tuple[T, ...]
 462    elif is_variable_tuple(typ):
 463        # Get the first type arg. Since tuple[T, ...] is homogeneous tuple,
 464        # all the elements should be of this type.
 465        arg = type_args(typ)[0]
 466        for v in obj:
 467            if not is_instance(v, arg):
 468                return False
 469        return True
 470
 471    # bare tuple "tuple" is equivalent to tuple[Any, ...]
 472    if is_bare_tuple(typ) and isinstance(obj, tuple):
 473        return True
 474
 475    # All the other tuples e.g. tuple[int, str]
 476    if len(obj) == len(args):
 477        for element, arg in zip(obj, args, strict=False):
 478            if not is_instance(element, arg):
 479                return False
 480    else:
 481        return False
 482
 483    return True
 484
 485
 486def is_dict_instance(obj: Any, typ: type[Any]) -> bool:
 487    if not isinstance(obj, Mapping):
 488        return False
 489    if len(obj) == 0 or is_bare_dict(typ):
 490        return True
 491    ktyp = type_args(typ)[0]
 492    vtyp = type_args(typ)[1]
 493    for k, v in obj.items():
 494        # for speed reasons we just check the type of the 1st element
 495        return is_instance(k, ktyp) and is_instance(v, vtyp)
 496    return False
 497
 498
 499def is_deque_instance(obj: Any, typ: type[Any]) -> bool:
 500    if not isinstance(obj, deque):
 501        return False
 502    if len(obj) == 0 or is_bare_deque(typ):
 503        return True
 504    deque_arg = type_args(typ)[0]
 505    # for speed reasons we just check the type of the 1st element
 506    return is_instance(obj[0], deque_arg)
 507
 508
 509def is_counter_instance(obj: Any, typ: type[Any]) -> bool:
 510    if not isinstance(obj, Counter):
 511        return False
 512    if len(obj) == 0 or is_bare_counter(typ):
 513        return True
 514    counter_arg = type_args(typ)[0]
 515    # for speed reasons we just check the type of the 1st key
 516    return is_instance(next(iter(obj.keys())), counter_arg)
 517
 518
 519def is_generic_instance(obj: Any, typ: type[Any]) -> bool:
 520    return is_instance(obj, get_origin(typ))
 521
 522
 523@dataclass
 524class Func:
 525    """
 526    Function wrapper that provides `mangled` optional field.
 527
 528    pyserde copies every function reference into global scope
 529    for code generation. Mangling function names is needed in
 530    order to avoid name conflict in the global scope when
 531    multiple fields receives `skip_if` attribute.
 532    """
 533
 534    inner: Callable[..., Any]
 535    """ Function to wrap in """
 536
 537    mangeld: str = ""
 538    """ Mangled function name """
 539
 540    def __call__(self, v: Any) -> None:
 541        return self.inner(v)  # type: ignore
 542
 543    @property
 544    def name(self) -> str:
 545        """
 546        Mangled function name
 547        """
 548        return self.mangeld
 549
 550
 551def skip_if_false(v: Any) -> Any:
 552    return not bool(v)
 553
 554
 555def skip_if_default(v: Any, default: Any | None = None) -> Any:
 556    return v == default  # Why return type is deduced to be Any?
 557
 558
 559@dataclass
 560class FlattenOpts:
 561    """
 562    Flatten options. Currently not used.
 563    """
 564
 565
 566def field(
 567    *args: Any,
 568    rename: str | None = None,
 569    alias: list[str] | None = None,
 570    skip: bool | None = None,
 571    skip_if: Callable[[Any], Any] | None = None,
 572    skip_if_false: bool | None = None,
 573    skip_if_default: bool | None = None,
 574    serializer: Callable[..., Any] | None = None,
 575    deserializer: Callable[..., Any] | None = None,
 576    flatten: FlattenOpts | bool | None = None,
 577    metadata: dict[str, Any] | None = None,
 578    **kwargs: Any,
 579) -> Any:
 580    """
 581    Declare a field with parameters.
 582    """
 583    if not metadata:
 584        metadata = {}
 585
 586    if rename is not None:
 587        metadata["serde_rename"] = rename
 588    if alias is not None:
 589        metadata["serde_alias"] = alias
 590    if skip is not None:
 591        metadata["serde_skip"] = skip
 592    if skip_if is not None:
 593        metadata["serde_skip_if"] = skip_if
 594    if skip_if_false is not None:
 595        metadata["serde_skip_if_false"] = skip_if_false
 596    if skip_if_default is not None:
 597        metadata["serde_skip_if_default"] = skip_if_default
 598    if serializer:
 599        metadata["serde_serializer"] = serializer
 600    if deserializer:
 601        metadata["serde_deserializer"] = deserializer
 602    if flatten is True:
 603        metadata["serde_flatten"] = FlattenOpts()
 604    elif flatten:
 605        metadata["serde_flatten"] = flatten
 606
 607    return dataclasses.field(*args, metadata=metadata, **kwargs)
 608
 609
 610@dataclass
 611class Field(Generic[T]):
 612    """
 613    Field class is similar to `dataclasses.Field`. It provides pyserde specific options.
 614
 615    `type`, `name`, `default` and `default_factory` are the same members as `dataclasses.Field`.
 616    """
 617
 618    type: type[T]
 619    """ Type of Field """
 620    name: str | None
 621    """ Name of Field """
 622    default: Any = field(default_factory=dataclasses._MISSING_TYPE)
 623    """ Default value of Field """
 624    default_factory: Any = field(default_factory=dataclasses._MISSING_TYPE)
 625    """ Default factory method of Field """
 626    init: bool = field(default_factory=dataclasses._MISSING_TYPE)
 627    repr: Any = field(default_factory=dataclasses._MISSING_TYPE)
 628    hash: Any = field(default_factory=dataclasses._MISSING_TYPE)
 629    compare: Any = field(default_factory=dataclasses._MISSING_TYPE)
 630    metadata: Mapping[str, Any] = field(default_factory=dict)
 631    kw_only: bool = False
 632    case: str | None = None
 633    alias: list[str] = field(default_factory=list)
 634    rename: str | None = None
 635    skip: bool | None = None
 636    skip_if: Func | None = None
 637    skip_if_false: bool | None = None
 638    skip_if_default: bool | None = None
 639    serializer: Func | None = None  # Custom field serializer.
 640    deserializer: Func | None = None  # Custom field deserializer.
 641    flatten: FlattenOpts | None = None
 642    parent: Any | None = None
 643    type_args: list[str] | None = None
 644
 645    @classmethod
 646    def from_dataclass(cls, f: dataclasses.Field[T], parent: Any | None = None) -> Field[T]:
 647        """
 648        Create `Field` object from `dataclasses.Field`.
 649        """
 650        skip_if_false_func: Func | None = None
 651        if f.metadata.get("serde_skip_if_false"):
 652            skip_if_false_func = Func(skip_if_false, cls.mangle(f, "skip_if_false"))
 653
 654        skip_if_default_func: Func | None = None
 655        if f.metadata.get("serde_skip_if_default"):
 656            skip_if_def = functools.partial(skip_if_default, default=f.default)
 657            skip_if_default_func = Func(skip_if_def, cls.mangle(f, "skip_if_default"))
 658
 659        skip_if: Func | None = None
 660        if f.metadata.get("serde_skip_if"):
 661            func = f.metadata.get("serde_skip_if")
 662            if callable(func):
 663                skip_if = Func(func, cls.mangle(f, "skip_if"))
 664
 665        serializer: Func | None = None
 666        func = f.metadata.get("serde_serializer")
 667        if func:
 668            serializer = Func(func, cls.mangle(f, "serializer"))
 669
 670        deserializer: Func | None = None
 671        func = f.metadata.get("serde_deserializer")
 672        if func:
 673            deserializer = Func(func, cls.mangle(f, "deserializer"))
 674
 675        flatten = f.metadata.get("serde_flatten")
 676        if flatten is True:
 677            flatten = FlattenOpts()
 678        if flatten:
 679            is_valid_flatten_type = (
 680                dataclasses.is_dataclass(f.type)
 681                or is_opt_dataclass(f.type)
 682                or is_flatten_dict(f.type)
 683            )
 684            if not is_valid_flatten_type:
 685                raise SerdeError(
 686                    f"pyserde does not support flatten attribute for {typename(f.type)}. "
 687                    "Supported types: dataclass, Optional[dataclass], dict[str, Any], dict"
 688                )
 689
 690        kw_only = bool(f.kw_only)
 691
 692        return cls(
 693            f.type,  # type: ignore
 694            f.name,
 695            default=f.default,
 696            default_factory=f.default_factory,
 697            init=f.init,
 698            repr=f.repr,
 699            hash=f.hash,
 700            compare=f.compare,
 701            metadata=f.metadata,
 702            rename=f.metadata.get("serde_rename"),
 703            alias=f.metadata.get("serde_alias", []),
 704            skip=f.metadata.get("serde_skip"),
 705            skip_if=skip_if or skip_if_false_func or skip_if_default_func,
 706            serializer=serializer,
 707            deserializer=deserializer,
 708            flatten=flatten,
 709            parent=parent,
 710            kw_only=kw_only,
 711        )
 712
 713    def to_dataclass(self) -> dataclasses.Field[T]:
 714        base_kwargs = {
 715            "init": self.init,
 716            "repr": self.repr,
 717            "hash": self.hash,
 718            "compare": self.compare,
 719            "metadata": self.metadata,
 720            "kw_only": self.kw_only,
 721            "default": self.default,
 722            "default_factory": self.default_factory,
 723        }
 724
 725        f = cast(dataclasses.Field[T], dataclasses.field(**base_kwargs))
 726        assert self.name
 727        f.name = self.name
 728        f.type = self.type
 729        return f
 730
 731    def is_self_referencing(self) -> bool:
 732        if self.type is None:
 733            return False
 734        if self.parent is None:
 735            return False
 736        return self.type == self.parent  # type: ignore
 737
 738    @staticmethod
 739    def mangle(field: dataclasses.Field[Any], name: str) -> str:
 740        """
 741        Get mangled name based on field name.
 742        """
 743        return f"{field.name}_{name}"
 744
 745    def conv_name(self, case: str | None = None) -> str:
 746        """
 747        Get an actual field name which `rename` and `rename_all` conversions
 748        are made. Use `name` property to get a field name before conversion.
 749        """
 750        return conv(self, case or self.case)
 751
 752    def supports_default(self) -> bool:
 753        return not getattr(self, "iterbased", False) and (
 754            has_default(self) or has_default_factory(self)
 755        )
 756
 757
 758F = TypeVar("F", bound=Field[Any])
 759
 760
 761def fields(field_cls: type[F], cls: type[Any], serialize_class_var: bool = False) -> list[F]:
 762    """
 763    Iterate fields of the dataclass and returns `serde.core.Field`.
 764    """
 765    fields = [field_cls.from_dataclass(f, parent=cls) for f in dataclass_fields(cls)]
 766
 767    if serialize_class_var:
 768        for name, typ in get_type_hints(cls).items():
 769            if is_class_var(typ):
 770                fields.append(field_cls(typ, name, default=getattr(cls, name)))
 771
 772    return fields  # type: ignore
 773
 774
 775def get_transparent_field(cls: type[Any]) -> Field[Any]:
 776    """
 777    Return the single "transparent" field for `cls`.
 778
 779    A transparent class must have exactly one init=True field that is not skipped.
 780    Any other dataclass fields must be both init=False and skipped.
 781    """
 782    all_fields: list[Field[Any]] = fields(Field, cls)
 783
 784    candidates = [f for f in all_fields if f.init and not f.skip]
 785    if len(candidates) != 1:
 786        raise SerdeError(
 787            f"{typename(cls)} with `transparent=True` must have exactly one init=True, "
 788            "non-skipped field"
 789        )
 790
 791    chosen = candidates[0]
 792    for f in all_fields:
 793        if f.name == chosen.name:
 794            continue
 795        if f.init:
 796            raise SerdeError(
 797                f"{typename(cls)} with `transparent=True` can not have additional init=True fields "
 798                f"(found {f.name!r})"
 799            )
 800        if not f.skip:
 801            raise SerdeError(
 802                f"{typename(cls)} with `transparent=True` requires non-transparent fields to be "
 803                f"skipped (set `serde.field(skip=True)`) (found {f.name!r})"
 804            )
 805
 806    return chosen
 807
 808
 809def conv(f: Field[Any], case: str | None = None) -> str:
 810    """
 811    Convert field name.
 812    """
 813    name = f.name
 814    if case:
 815        casef = getattr(casefy, case, None)
 816        if not casef:
 817            raise SerdeError(
 818                f"Unkown case type: {f.case}. Pass the name of case supported by 'casefy' package."
 819            )
 820        name = casef(name)
 821    if f.rename:
 822        name = f.rename
 823    if name is None:
 824        raise SerdeError("Field name is None.")
 825    return name
 826
 827
 828def union_func_name(prefix: str, union_args: Sequence[Any]) -> str:
 829    """
 830    Generate a function name that contains all union types
 831
 832    * `prefix` prefix to distinguish between serializing and deserializing
 833    * `union_args`: type arguments of a Union
 834
 835    >>> from ipaddress import IPv4Address
 836    >>> union_func_name("union_se", [int, list[str], IPv4Address])
 837    'union_se_int_list_str__IPv4Address'
 838    """
 839    return re.sub(r"[^A-Za-z0-9]", "_", f"{prefix}_{'_'.join([typename(e) for e in union_args])}")
 840
 841
 842def literal_func_name(literal_args: Sequence[Any]) -> str:
 843    """
 844    Generate a function name with all literals and corresponding types specified with Literal[...]
 845
 846
 847    * `literal_args`: arguments of a Literal
 848
 849    >>> literal_func_name(["r", "w", "a", "x", "r+", "w+", "a+", "x+"])
 850    'literal_de_r_str_w_str_a_str_x_str_r__str_w__str_a__str_x__str'
 851    """
 852    return re.sub(
 853        r"[^A-Za-z0-9]",
 854        "_",
 855        f"{LITERAL_DE_PREFIX}_{'_'.join(f'{a}_{typename(type(a))}' for a in literal_args)}",
 856    )
 857
 858
 859@dataclass(frozen=True)
 860class Tagging:
 861    """
 862    Controls how union is (de)serialized. This is the same concept as in
 863    https://serde.rs/enum-representations.html
 864    """
 865
 866    class Kind(enum.Enum):
 867        External = enum.auto()
 868        Internal = enum.auto()
 869        Adjacent = enum.auto()
 870        Untagged = enum.auto()
 871
 872    tag: str | None = None
 873    content: str | None = None
 874    kind: Kind = Kind.External
 875
 876    def is_external(self) -> bool:
 877        return self.kind == self.Kind.External
 878
 879    def is_internal(self) -> bool:
 880        return self.kind == self.Kind.Internal
 881
 882    def is_adjacent(self) -> bool:
 883        return self.kind == self.Kind.Adjacent
 884
 885    def is_untagged(self) -> bool:
 886        return self.kind == self.Kind.Untagged
 887
 888    @classmethod
 889    def is_taggable(cls, typ: type[Any]) -> bool:
 890        return dataclasses.is_dataclass(typ)
 891
 892    def check(self) -> None:
 893        if self.is_internal() and self.tag is None:
 894            raise SerdeError('"tag" must be specified in InternalTagging')
 895        if self.is_adjacent() and (self.tag is None or self.content is None):
 896            raise SerdeError('"tag" and "content" must be specified in AdjacentTagging')
 897
 898    def produce_unique_class_name(self) -> str:
 899        """
 900        Produce a unique class name for this tagging. The name is used for generated
 901        wrapper dataclass and stored in `Cache`.
 902        """
 903        if self.is_internal():
 904            tag = casefy.pascalcase(self.tag)  # type: ignore
 905            if not tag:
 906                raise SerdeError('"tag" must be specified in InternalTagging')
 907            return f"Internal{tag}"
 908        elif self.is_adjacent():
 909            tag = casefy.pascalcase(self.tag)  # type: ignore
 910            content = casefy.pascalcase(self.content)  # type: ignore
 911            if not tag:
 912                raise SerdeError('"tag" must be specified in AdjacentTagging')
 913            if not content:
 914                raise SerdeError('"content" must be specified in AdjacentTagging')
 915            return f"Adjacent{tag}{content}"
 916        else:
 917            return self.kind.name
 918
 919    def __call__(self, cls: T) -> _WithTagging[T]:
 920        return _WithTagging(cls, self)
 921
 922
 923@overload
 924def InternalTagging(tag: str) -> Tagging: ...
 925
 926
 927@overload
 928def InternalTagging(tag: str, cls: T) -> _WithTagging[T]: ...
 929
 930
 931def InternalTagging(tag: str, cls: T | None = None) -> Tagging | _WithTagging[T]:
 932    tagging = Tagging(tag, kind=Tagging.Kind.Internal)
 933    if cls:
 934        return tagging(cls)
 935    else:
 936        return tagging
 937
 938
 939@overload
 940def AdjacentTagging(tag: str, content: str) -> Tagging: ...
 941
 942
 943@overload
 944def AdjacentTagging(tag: str, content: str, cls: T) -> _WithTagging[T]: ...
 945
 946
 947def AdjacentTagging(tag: str, content: str, cls: T | None = None) -> Tagging | _WithTagging[T]:
 948    tagging = Tagging(tag, content, kind=Tagging.Kind.Adjacent)
 949    if cls:
 950        return tagging(cls)
 951    else:
 952        return tagging
 953
 954
 955ExternalTagging = Tagging()
 956
 957Untagged = Tagging(kind=Tagging.Kind.Untagged)
 958
 959
 960DefaultTagging = ExternalTagging
 961
 962
 963def ensure(expr: Any, description: str) -> None:
 964    if not expr:
 965        raise Exception(description)
 966
 967
 968def should_impl_dataclass(cls: type[Any]) -> bool:
 969    """
 970    Test if class doesn't have @dataclass.
 971
 972    `dataclasses.is_dataclass` returns True even Derived class doesn't actually @dataclass.
 973    >>> @dataclasses.dataclass
 974    ... class Base:
 975    ...     a: int
 976    >>> class Derived(Base):
 977    ...     b: int
 978    >>> dataclasses.is_dataclass(Derived)
 979    True
 980
 981    This function tells the class actually have @dataclass or not.
 982    >>> should_impl_dataclass(Base)
 983    False
 984    >>> should_impl_dataclass(Derived)
 985    True
 986    """
 987    if not dataclasses.is_dataclass(cls):
 988        return True
 989
 990    # Checking is_dataclass is not enough in such a case that the class is inherited
 991    # from another dataclass. To do it correctly, check if all fields in __annotations__
 992    # are present as dataclass fields.
 993    annotations = getattr(cls, "__annotations__", {})
 994    if not annotations:
 995        return False
 996
 997    field_names = [field.name for field in dataclass_fields(cls)]
 998    for field_name, annotation in annotations.items():
 999        # Omit InitVar field because it doesn't appear in dataclass fields.
1000        if is_instance(annotation, dataclasses.InitVar):
1001            continue
1002        # This field in __annotations__ is not a part of dataclass fields.
1003        # This means this class does not implement dataclass directly.
1004        if field_name not in field_names:
1005            return True
1006
1007    # If all of the fields in __annotation__ are present as dataclass fields,
1008    # the class already implemented dataclass, thus returns False.
1009    return False
1010
1011
1012@dataclass
1013class TypeCheck:
1014    """
1015    Specify type check flavors.
1016    """
1017
1018    class Kind(enum.Enum):
1019        Disabled = enum.auto()
1020        """ No check performed """
1021
1022        Coerce = enum.auto()
1023        """ Value is coerced into the declared type """
1024        Strict = enum.auto()
1025        """ Value are strictly checked against the declared type """
1026
1027    kind: Kind
1028
1029    def is_strict(self) -> bool:
1030        return self.kind == self.Kind.Strict
1031
1032    def is_coerce(self) -> bool:
1033        return self.kind == self.Kind.Coerce
1034
1035    def __call__(self, **kwargs: Any) -> TypeCheck:
1036        # TODO
1037        return self
1038
1039
1040disabled = TypeCheck(kind=TypeCheck.Kind.Disabled)
1041
1042coerce = TypeCheck(kind=TypeCheck.Kind.Coerce)
1043
1044strict = TypeCheck(kind=TypeCheck.Kind.Strict)
1045
1046
1047def coerce_object(cls: str, field: str, typ: type[Any], obj: Any) -> Any:
1048    try:
1049        return typ(obj) if is_coercible(typ, obj) else obj
1050    except Exception as e:
1051        raise SerdeError(
1052            f"failed to coerce the field {cls}.{field} value {obj} into {typename(typ)}: {e}"
1053        )
1054
1055
1056def is_coercible(typ: type[Any], obj: Any) -> bool:
1057    if obj is None:
1058        return False
1059    return True
1060
1061
1062def has_default(field: Field[Any]) -> bool:
1063    """
1064    Test if the field has default value.
1065
1066    >>> @dataclasses.dataclass
1067    ... class C:
1068    ...     a: int
1069    ...     d: int = 10
1070    >>> has_default(dataclasses.fields(C)[0])
1071    False
1072    >>> has_default(dataclasses.fields(C)[1])
1073    True
1074    """
1075    return not isinstance(field.default, dataclasses._MISSING_TYPE)
1076
1077
1078def has_default_factory(field: Field[Any]) -> bool:
1079    """
1080    Test if the field has default factory.
1081
1082    >>> @dataclasses.dataclass
1083    ... class C:
1084    ...     a: int
1085    ...     d: dict = dataclasses.field(default_factory=dict)
1086    >>> has_default_factory(dataclasses.fields(C)[0])
1087    False
1088    >>> has_default_factory(dataclasses.fields(C)[1])
1089    True
1090    """
1091    return not isinstance(field.default_factory, dataclasses._MISSING_TYPE)
1092
1093
1094class ClassSerializer(Protocol):
1095    """
1096    Interface for custom class serializer.
1097
1098    This protocol is intended to be used for custom class serializer.
1099
1100    >>> from datetime import datetime
1101    >>> from serde import serde
1102    >>> from plum import dispatch
1103    >>> class MySerializer(ClassSerializer):
1104    ...     @dispatch
1105    ...     def serialize(self, value: datetime) -> str:
1106    ...         return value.strftime("%d/%m/%y")
1107    """
1108
1109    def serialize(self, value: Any) -> Any:
1110        pass
1111
1112
1113class ClassDeserializer(Protocol):
1114    """
1115    Interface for custom class deserializer.
1116
1117    This protocol is intended to be used for custom class deserializer.
1118
1119    >>> from datetime import datetime
1120    >>> from serde import serde
1121    >>> from plum import dispatch
1122    >>> class MyDeserializer(ClassDeserializer):
1123    ...     @dispatch
1124    ...     def deserialize(self, cls: type[datetime], value: Any) -> datetime:
1125    ...         return datetime.strptime(value, "%d/%m/%y")
1126    """
1127
1128    def deserialize(self, cls: Any, value: Any) -> Any:
1129        pass
1130
1131
1132GLOBAL_CLASS_SERIALIZER: list[ClassSerializer] = []
1133
1134GLOBAL_CLASS_DESERIALIZER: list[ClassDeserializer] = []
1135
1136
1137def add_serializer(serializer: ClassSerializer) -> None:
1138    """
1139    Register custom global serializer.
1140    """
1141    GLOBAL_CLASS_SERIALIZER.append(serializer)
1142
1143
1144def add_deserializer(deserializer: ClassDeserializer) -> None:
1145    """
1146    Register custom global deserializer.
1147    """
1148    GLOBAL_CLASS_DESERIALIZER.append(deserializer)