_serializers.py 1.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051
  1. from __future__ import annotations
  2. import collections
  3. import collections.abc
  4. import typing
  5. from typing import Any
  6. from pydantic_core import PydanticOmit, core_schema
  7. SEQUENCE_ORIGIN_MAP: dict[Any, Any] = {
  8. typing.Deque: collections.deque,
  9. collections.deque: collections.deque,
  10. list: list,
  11. typing.List: list,
  12. set: set,
  13. typing.AbstractSet: set,
  14. typing.Set: set,
  15. frozenset: frozenset,
  16. typing.FrozenSet: frozenset,
  17. typing.Sequence: list,
  18. typing.MutableSequence: list,
  19. typing.MutableSet: set,
  20. # this doesn't handle subclasses of these
  21. # parametrized typing.Set creates one of these
  22. collections.abc.MutableSet: set,
  23. collections.abc.Set: frozenset,
  24. }
  25. def serialize_sequence_via_list(
  26. v: Any, handler: core_schema.SerializerFunctionWrapHandler, info: core_schema.SerializationInfo
  27. ) -> Any:
  28. items: list[Any] = []
  29. mapped_origin = SEQUENCE_ORIGIN_MAP.get(type(v), None)
  30. if mapped_origin is None:
  31. # we shouldn't hit this branch, should probably add a serialization error or something
  32. return v
  33. for index, item in enumerate(v):
  34. try:
  35. v = handler(item, index)
  36. except PydanticOmit:
  37. pass
  38. else:
  39. items.append(v)
  40. if info.mode_is_json():
  41. return items
  42. else:
  43. return mapped_origin(items)