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