Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/attr/_compat.py: 43%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# SPDX-License-Identifier: MIT
3import sys
4import threading
6from collections.abc import Mapping, Sequence # noqa: F401
7from typing import Callable, _GenericAlias
10PYPY = sys.implementation.name == "pypy"
11PY_3_11_PLUS = sys.version_info[:2] >= (3, 11)
12PY_3_12_PLUS = sys.version_info[:2] >= (3, 12)
13PY_3_13_PLUS = sys.version_info[:2] >= (3, 13)
14PY_3_14_PLUS = sys.version_info[:2] >= (3, 14)
17if PY_3_14_PLUS:
18 # We request forward-ref annotations to not break in the presence of
19 # forward references.
21 def _get_annotations(cls):
22 import annotationlib
24 return annotationlib.get_annotations(
25 cls, format=annotationlib.Format.FORWARDREF
26 )
28else:
30 def _get_annotations(cls):
31 """
32 Get annotations for *cls*.
33 """
34 return cls.__dict__.get("__annotations__", {}) # noqa: RUF063
37class _AnnotationExtractor:
38 """
39 Extract type annotations from a callable, returning None whenever there
40 is none.
41 """
43 __slots__ = ["sig"]
45 def __init__(self, callable):
46 import inspect
48 try:
49 self.sig = inspect.signature(callable)
50 except (ValueError, TypeError): # inspect failed
51 self.sig = None
53 def get_first_param_type(self):
54 """
55 Return the type annotation of the first argument if it's not empty.
56 """
57 import inspect
59 if not self.sig:
60 return None
62 params = list(self.sig.parameters.values())
63 if params and params[0].annotation is not inspect.Parameter.empty:
64 return params[0].annotation
66 return None
68 def get_return_type(self):
69 """
70 Return the return type if it's not empty.
71 """
72 import inspect
74 if (
75 self.sig
76 and self.sig.return_annotation is not inspect.Signature.empty
77 ):
78 return self.sig.return_annotation
80 return None
83# Thread-local global to track attrs instances which are already being repr'd.
84# This is needed because there is no other (thread-safe) way to pass info
85# about the instances that are already being repr'd through the call stack
86# in order to ensure we don't perform infinite recursion.
87#
88# For instance, if an instance contains a dict which contains that instance,
89# we need to know that we're already repr'ing the outside instance from within
90# the dict's repr() call.
91#
92# This lives here rather than in _make.py so that the functions in _make.py
93# don't have a direct reference to the thread-local in their globals dict.
94# If they have such a reference, it breaks cloudpickle.
95repr_context = threading.local()
98def get_generic_base(cl):
99 """If this is a generic class (A[str]), return the generic base for it."""
100 if cl.__class__ is _GenericAlias:
101 return cl.__origin__
102 return None
105_IS_GENERATOR_RESULTS = {}
108def _lazy_is_generator(f: Callable) -> Callable[[], bool]:
109 """
110 Return a caching closure over callable f that returns whether f is a
111 generator function.
113 Not thread-safe but doesn't matter.
114 """
116 def is_gen() -> bool:
117 import inspect
119 try:
120 if f not in _IS_GENERATOR_RESULTS:
121 _IS_GENERATOR_RESULTS[f] = inspect.isgeneratorfunction(f)
122 except TypeError: # f is not hashable
123 return inspect.isgeneratorfunction(f)
125 return _IS_GENERATOR_RESULTS[f]
127 return is_gen