Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/wrapt/weakrefs.py: 22%

1"""Weak reference proxy for functions and bound methods.""" 

2 

3import functools 

4import weakref 

5 

6from .__wrapt__ import BaseObjectProxy, _FunctionWrapperBase 

7 

8# A weak function proxy. This will work on instance methods, class 

9# methods, static methods and regular functions. Special treatment is 

10# needed for the method types because the bound method is effectively a 

11# transient object and applying a weak reference to one will immediately 

12# result in it being destroyed and the weakref callback called. The weak 

13# reference is therefore applied to the instance the method is bound to 

14# and the original function. The function is then rebound at the point 

15# of a call via the weak function proxy. 

16 

17 

18def _weak_function_proxy_callback(ref, proxy, callback): 

19 if proxy._self_expired: 

20 return 

21 

22 proxy._self_expired = True 

23 

24 # This could raise an exception. We let it propagate back and let 

25 # the weakref.proxy() deal with it, at which point it generally 

26 # prints out a short error message direct to stderr and keeps going. 

27 

28 if callback is not None: 

29 callback(proxy) 

30 

31 

32class WeakFunctionProxy(BaseObjectProxy): 

33 """A weak function proxy.""" 

34 

35 def __init__(self, wrapped, callback=None): 

36 """Create a proxy to object which uses a weak reference. This is 

37 similar to the `weakref.proxy` but is designed to work with functions 

38 and methods. It will automatically rebind the function to the instance 

39 when called if the function was originally a bound method. This is 

40 necessary because bound methods are transient objects and applying a 

41 weak reference to one will immediately result in it being destroyed 

42 and the weakref callback called. The weak reference is therefore 

43 applied to the instance the method is bound to and the original 

44 function. The function is then rebound at the point of a call via the 

45 weak function proxy. 

46 """ 

47 

48 # We need to determine if the wrapped function is actually a 

49 # bound method. In the case of a bound method, we need to keep a 

50 # reference to the original unbound function and the instance. 

51 # This is necessary because if we hold a reference to the bound 

52 # function, it will be the only reference and given it is a 

53 # temporary object, it will almost immediately expire and 

54 # the weakref callback triggered. So what is done is that we 

55 # hold a reference to the instance and unbound function and 

56 # when called bind the function to the instance once again and 

57 # then call it. Note that we avoid using a nested function for 

58 # the callback here so as not to cause any odd reference cycles. 

59 

60 _callback = callback and functools.partial( 

61 _weak_function_proxy_callback, proxy=self, callback=callback 

62 ) 

63 

64 self._self_expired = False 

65 

66 if isinstance(wrapped, _FunctionWrapperBase): 

67 self._self_instance = weakref.ref(wrapped._self_instance, _callback) 

68 

69 if wrapped._self_parent is not None: 

70 # Explicit class in super() is used because the proxy 

71 # overrides __class__ and MRO-related methods to delegate 

72 # to the wrapped object, which can interfere with bare 

73 # super(). 

74 super(WeakFunctionProxy, self).__init__( 

75 weakref.proxy(wrapped._self_parent, _callback) 

76 ) 

77 

78 else: 

79 super(WeakFunctionProxy, self).__init__( 

80 weakref.proxy(wrapped, _callback) 

81 ) 

82 

83 return 

84 

85 try: 

86 self._self_instance = weakref.ref(wrapped.__self__, _callback) 

87 

88 super(WeakFunctionProxy, self).__init__( 

89 weakref.proxy(wrapped.__func__, _callback) 

90 ) 

91 

92 except AttributeError: 

93 self._self_instance = None 

94 

95 super(WeakFunctionProxy, self).__init__(weakref.proxy(wrapped, _callback)) 

96 

97 def __call__(*args, **kwargs): 

98 def _unpack_self(self, *args): 

99 return self, args 

100 

101 self, args = _unpack_self(*args) 

102 

103 # We perform a boolean check here on the instance and wrapped 

104 # function as that will trigger the reference error prior to 

105 # calling if the reference had expired. 

106 

107 instance = self._self_instance and self._self_instance() 

108 function = self.__wrapped__ and self.__wrapped__ 

109 

110 # If the wrapped function was originally a bound method but the 

111 # instance it was bound to has been garbage collected, raise a 

112 # ReferenceError rather than silently calling it as unbound. 

113 

114 if self._self_instance is not None and instance is None: 

115 raise ReferenceError("weakly-referenced object no longer exists") 

116 

117 # If the wrapped function was originally a bound function, for 

118 # which we retained a reference to the instance and the unbound 

119 # function we need to rebind the function and then call it. If 

120 # not just called the wrapped function. 

121 

122 if instance is None: 

123 return self.__wrapped__(*args, **kwargs) 

124 

125 return function.__get__(instance, type(instance))(*args, **kwargs)