Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/blinker/_utilities.py: 45%

1from __future__ import annotations 

2 

3import asyncio 

4import inspect 

5import sys 

6import typing as t 

7from functools import partial 

8from weakref import ref 

9 

10from blinker._saferef import BoundMethodWeakref 

11 

12IdentityType = t.Union[t.Tuple[int, int], str, int] 

13 

14 

15class _symbol: 

16 def __init__(self, name): 

17 """Construct a new named symbol.""" 

18 self.__name__ = self.name = name 

19 

20 def __reduce__(self): 

21 return symbol, (self.name,) 

22 

23 def __repr__(self): 

24 return self.name 

25 

26 

27_symbol.__name__ = "symbol" 

28 

29 

30class symbol: 

31 """A constant symbol. 

32 

33 >>> symbol('foo') is symbol('foo') 

34 True 

35 >>> symbol('foo') 

36 foo 

37 

38 A slight refinement of the MAGICCOOKIE=object() pattern. The primary 

39 advantage of symbol() is its repr(). They are also singletons. 

40 

41 Repeated calls of symbol('name') will all return the same instance. 

42 

43 """ 

44 

45 symbols = {} # type: ignore[var-annotated] 

46 

47 def __new__(cls, name): 

48 try: 

49 return cls.symbols[name] 

50 except KeyError: 

51 return cls.symbols.setdefault(name, _symbol(name)) 

52 

53 

54def hashable_identity(obj: object) -> IdentityType: 

55 if hasattr(obj, "__func__"): 

56 return (id(obj.__func__), id(obj.__self__)) # type: ignore[attr-defined] 

57 elif hasattr(obj, "im_func"): 

58 return (id(obj.im_func), id(obj.im_self)) # type: ignore[attr-defined] 

59 elif isinstance(obj, (int, str)): 

60 return obj 

61 else: 

62 return id(obj) 

63 

64 

65WeakTypes = (ref, BoundMethodWeakref) 

66 

67 

68class annotatable_weakref(ref): 

69 """A weakref.ref that supports custom instance attributes.""" 

70 

71 receiver_id: t.Optional[IdentityType] 

72 sender_id: t.Optional[IdentityType] 

73 

74 

75def reference( # type: ignore[no-untyped-def] 

76 object, callback=None, **annotations 

77) -> annotatable_weakref: 

78 """Return an annotated weak ref.""" 

79 if callable(object): 

80 weak = callable_reference(object, callback) 

81 else: 

82 weak = annotatable_weakref(object, callback) 

83 for key, value in annotations.items(): 

84 setattr(weak, key, value) 

85 return weak # type: ignore[no-any-return] 

86 

87 

88def callable_reference(object, callback=None): 

89 """Return an annotated weak ref, supporting bound instance methods.""" 

90 if hasattr(object, "im_self") and object.im_self is not None: 

91 return BoundMethodWeakref(target=object, on_delete=callback) 

92 elif hasattr(object, "__self__") and object.__self__ is not None: 

93 return BoundMethodWeakref(target=object, on_delete=callback) 

94 return annotatable_weakref(object, callback) 

95 

96 

97class lazy_property: 

98 """A @property that is only evaluated once.""" 

99 

100 def __init__(self, deferred): 

101 self._deferred = deferred 

102 self.__doc__ = deferred.__doc__ 

103 

104 def __get__(self, obj, cls): 

105 if obj is None: 

106 return self 

107 value = self._deferred(obj) 

108 setattr(obj, self._deferred.__name__, value) 

109 return value 

110 

111 

112def is_coroutine_function(func: t.Any) -> bool: 

113 # Python < 3.8 does not correctly determine partially wrapped 

114 # coroutine functions are coroutine functions, hence the need for 

115 # this to exist. Code taken from CPython. 

116 if sys.version_info >= (3, 8): 

117 return asyncio.iscoroutinefunction(func) 

118 else: 

119 # Note that there is something special about the AsyncMock 

120 # such that it isn't determined as a coroutine function 

121 # without an explicit check. 

122 try: 

123 from unittest.mock import AsyncMock # type: ignore[attr-defined] 

124 

125 if isinstance(func, AsyncMock): 

126 return True 

127 except ImportError: 

128 # Not testing, no asynctest to import 

129 pass 

130 

131 while inspect.ismethod(func): 

132 func = func.__func__ 

133 while isinstance(func, partial): 

134 func = func.func 

135 if not inspect.isfunction(func): 

136 return False 

137 

138 if func.__code__.co_flags & inspect.CO_COROUTINE: 

139 return True 

140 

141 acic = asyncio.coroutines._is_coroutine # type: ignore[attr-defined] 

142 return getattr(func, "_is_coroutine", None) is acic