_serializers.py 1.5 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253
  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, # noqa: UP006
  9. collections.deque: collections.deque,
  10. list: list,
  11. typing.List: list, # noqa: UP006
  12. tuple: tuple,
  13. typing.Tuple: tuple, # noqa: UP006
  14. set: set,
  15. typing.AbstractSet: set,
  16. typing.Set: set, # noqa: UP006
  17. frozenset: frozenset,
  18. typing.FrozenSet: frozenset, # noqa: UP006
  19. typing.Sequence: list,
  20. typing.MutableSequence: list,
  21. typing.MutableSet: set,
  22. # this doesn't handle subclasses of these
  23. # parametrized typing.Set creates one of these
  24. collections.abc.MutableSet: set,
  25. collections.abc.Set: frozenset,
  26. }
  27. def serialize_sequence_via_list(
  28. v: Any, handler: core_schema.SerializerFunctionWrapHandler, info: core_schema.SerializationInfo
  29. ) -> Any:
  30. items: list[Any] = []
  31. mapped_origin = SEQUENCE_ORIGIN_MAP.get(type(v), None)
  32. if mapped_origin is None:
  33. # we shouldn't hit this branch, should probably add a serialization error or something
  34. return v
  35. for index, item in enumerate(v):
  36. try:
  37. v = handler(item, index)
  38. except PydanticOmit: # noqa: PERF203
  39. pass
  40. else:
  41. items.append(v)
  42. if info.mode_is_json():
  43. return items
  44. else:
  45. return mapped_origin(items)