_compat.py 2.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899
  1. # SPDX-License-Identifier: MIT
  2. import inspect
  3. import platform
  4. import sys
  5. import threading
  6. from collections.abc import Mapping, Sequence # noqa: F401
  7. from typing import _GenericAlias
  8. PYPY = platform.python_implementation() == "PyPy"
  9. PY_3_10_PLUS = sys.version_info[:2] >= (3, 10)
  10. PY_3_11_PLUS = sys.version_info[:2] >= (3, 11)
  11. PY_3_12_PLUS = sys.version_info[:2] >= (3, 12)
  12. PY_3_13_PLUS = sys.version_info[:2] >= (3, 13)
  13. PY_3_14_PLUS = sys.version_info[:2] >= (3, 14)
  14. if PY_3_14_PLUS:
  15. import annotationlib
  16. # We request forward-ref annotations to not break in the presence of
  17. # forward references.
  18. def _get_annotations(cls):
  19. return annotationlib.get_annotations(
  20. cls, format=annotationlib.Format.FORWARDREF
  21. )
  22. else:
  23. def _get_annotations(cls):
  24. """
  25. Get annotations for *cls*.
  26. """
  27. return cls.__dict__.get("__annotations__", {})
  28. class _AnnotationExtractor:
  29. """
  30. Extract type annotations from a callable, returning None whenever there
  31. is none.
  32. """
  33. __slots__ = ["sig"]
  34. def __init__(self, callable):
  35. try:
  36. self.sig = inspect.signature(callable)
  37. except (ValueError, TypeError): # inspect failed
  38. self.sig = None
  39. def get_first_param_type(self):
  40. """
  41. Return the type annotation of the first argument if it's not empty.
  42. """
  43. if not self.sig:
  44. return None
  45. params = list(self.sig.parameters.values())
  46. if params and params[0].annotation is not inspect.Parameter.empty:
  47. return params[0].annotation
  48. return None
  49. def get_return_type(self):
  50. """
  51. Return the return type if it's not empty.
  52. """
  53. if (
  54. self.sig
  55. and self.sig.return_annotation is not inspect.Signature.empty
  56. ):
  57. return self.sig.return_annotation
  58. return None
  59. # Thread-local global to track attrs instances which are already being repr'd.
  60. # This is needed because there is no other (thread-safe) way to pass info
  61. # about the instances that are already being repr'd through the call stack
  62. # in order to ensure we don't perform infinite recursion.
  63. #
  64. # For instance, if an instance contains a dict which contains that instance,
  65. # we need to know that we're already repr'ing the outside instance from within
  66. # the dict's repr() call.
  67. #
  68. # This lives here rather than in _make.py so that the functions in _make.py
  69. # don't have a direct reference to the thread-local in their globals dict.
  70. # If they have such a reference, it breaks cloudpickle.
  71. repr_context = threading.local()
  72. def get_generic_base(cl):
  73. """If this is a generic class (A[str]), return the generic base for it."""
  74. if cl.__class__ is _GenericAlias:
  75. return cl.__origin__
  76. return None