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

1import functools 

2import weakref 

3 

4from .__wrapt__ import ObjectProxy, _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 

15def _weak_function_proxy_callback(ref, proxy, callback): 

16 if proxy._self_expired: 

17 return 

18 

19 proxy._self_expired = True 

20 

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

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

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

24 

25 if callback is not None: 

26 callback(proxy) 

27 

28class WeakFunctionProxy(ObjectProxy): 

29 

30 __slots__ = ('_self_expired', '_self_instance') 

31 

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

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

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

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

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

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

38 # temporary object, it will almost immediately expire and 

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

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

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

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

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

44 

45 _callback = callback and functools.partial( 

46 _weak_function_proxy_callback, proxy=self, 

47 callback=callback) 

48 

49 self._self_expired = False 

50 

51 if isinstance(wrapped, _FunctionWrapperBase): 

52 self._self_instance = weakref.ref(wrapped._self_instance, 

53 _callback) 

54 

55 if wrapped._self_parent is not None: 

56 super(WeakFunctionProxy, self).__init__( 

57 weakref.proxy(wrapped._self_parent, _callback)) 

58 

59 else: 

60 super(WeakFunctionProxy, self).__init__( 

61 weakref.proxy(wrapped, _callback)) 

62 

63 return 

64 

65 try: 

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

67 

68 super(WeakFunctionProxy, self).__init__( 

69 weakref.proxy(wrapped.__func__, _callback)) 

70 

71 except AttributeError: 

72 self._self_instance = None 

73 

74 super(WeakFunctionProxy, self).__init__( 

75 weakref.proxy(wrapped, _callback)) 

76 

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

78 def _unpack_self(self, *args): 

79 return self, args 

80 

81 self, args = _unpack_self(*args) 

82 

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

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

85 # calling if the reference had expired. 

86 

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

88 function = self.__wrapped__ and self.__wrapped__ 

89 

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

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

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

93 # not just called the wrapped function. 

94 

95 if instance is None: 

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

97 

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