_compat.py 2.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687
  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_8_PLUS = sys.version_info[:2] >= (3, 8)
  10. PY_3_9_PLUS = sys.version_info[:2] >= (3, 9)
  11. PY310 = sys.version_info[:2] >= (3, 10)
  12. PY_3_12_PLUS = sys.version_info[:2] >= (3, 12)
  13. if sys.version_info < (3, 8):
  14. try:
  15. from typing_extensions import Protocol
  16. except ImportError: # pragma: no cover
  17. Protocol = object
  18. else:
  19. from typing import Protocol # noqa: F401
  20. class _AnnotationExtractor:
  21. """
  22. Extract type annotations from a callable, returning None whenever there
  23. is none.
  24. """
  25. __slots__ = ["sig"]
  26. def __init__(self, callable):
  27. try:
  28. self.sig = inspect.signature(callable)
  29. except (ValueError, TypeError): # inspect failed
  30. self.sig = None
  31. def get_first_param_type(self):
  32. """
  33. Return the type annotation of the first argument if it's not empty.
  34. """
  35. if not self.sig:
  36. return None
  37. params = list(self.sig.parameters.values())
  38. if params and params[0].annotation is not inspect.Parameter.empty:
  39. return params[0].annotation
  40. return None
  41. def get_return_type(self):
  42. """
  43. Return the return type if it's not empty.
  44. """
  45. if (
  46. self.sig
  47. and self.sig.return_annotation is not inspect.Signature.empty
  48. ):
  49. return self.sig.return_annotation
  50. return None
  51. # Thread-local global to track attrs instances which are already being repr'd.
  52. # This is needed because there is no other (thread-safe) way to pass info
  53. # about the instances that are already being repr'd through the call stack
  54. # in order to ensure we don't perform infinite recursion.
  55. #
  56. # For instance, if an instance contains a dict which contains that instance,
  57. # we need to know that we're already repr'ing the outside instance from within
  58. # the dict's repr() call.
  59. #
  60. # This lives here rather than in _make.py so that the functions in _make.py
  61. # don't have a direct reference to the thread-local in their globals dict.
  62. # If they have such a reference, it breaks cloudpickle.
  63. repr_context = threading.local()
  64. def get_generic_base(cl):
  65. """If this is a generic class (A[str]), return the generic base for it."""
  66. if cl.__class__ is _GenericAlias:
  67. return cl.__origin__
  68. return None