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

1import functools 

2import weakref 

3 

4from .__wrapt__ import BaseObjectProxy, _FunctionWrapperBase 

5 

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

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

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

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

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

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

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

13# of a call via the weak function proxy. 

14 

15 

16def _weak_function_proxy_callback(ref, proxy, callback): 

17 if proxy._self_expired: 

18 return 

19 

20 proxy._self_expired = True 

21 

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

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

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

25 

26 if callback is not None: 

27 callback(proxy) 

28 

29 

30class WeakFunctionProxy(BaseObjectProxy): 

31 """A weak function proxy.""" 

32 

33 __slots__ = ("_self_expired", "_self_instance") 

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 super(WeakFunctionProxy, self).__init__( 

71 weakref.proxy(wrapped._self_parent, _callback) 

72 ) 

73 

74 else: 

75 super(WeakFunctionProxy, self).__init__( 

76 weakref.proxy(wrapped, _callback) 

77 ) 

78 

79 return 

80 

81 try: 

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

83 

84 super(WeakFunctionProxy, self).__init__( 

85 weakref.proxy(wrapped.__func__, _callback) 

86 ) 

87 

88 except AttributeError: 

89 self._self_instance = None 

90 

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

92 

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

94 def _unpack_self(self, *args): 

95 return self, args 

96 

97 self, args = _unpack_self(*args) 

98 

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

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

101 # calling if the reference had expired. 

102 

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

104 function = self.__wrapped__ and self.__wrapped__ 

105 

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

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

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

109 # not just called the wrapped function. 

110 

111 if instance is None: 

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

113 

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