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