public
anonymous / finalize.py
Created

Module for the finalization of weakrefable objects

  • Download Gist
finalize.py
Python
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156
'''
Module for the finalization of weakrefable objects
 
finalize(obj, func, *args, **kwds) creates a callable finalizer object
for obj which can be called explicitly, or automatically when obj is
garbage collected. For example
 
>>> class Kenny: pass
...
>>> kenny = Kenny()
>>> finalize(kenny, print, "you killed kenny!")
<finalize.finalize object at 0xffed4f2c>
>>> del kenny
you killed kenny!
 
Calling the finalizer for the first time will invoke
func(*args, **kwds) and return the result. Calling it a second
time has no effect and simply returns None.
 
>>> def callback():
... print("CALLBACK")
... return 123
...
>>> kenny = Kenny()
>>> f = finalize(kenny, callback)
>>> assert f() is 123
CALLBACK
>>> f() # no effect because finalizer already called
>>> del kenny # no effect because finalizer already called
 
At program shutdown, any remaining finalizers for which the atexit
attribute is true will be called in reverse order of creation. By
default the atexit attribute is false.
 
>>> kenny1 = Kenny()
>>> f1 = finalize(kenny1, print, "kenny1")
>>> kenny2 = Kenny()
>>> f2 = finalize(kenny2, print, "kenny2")
>>> f2.atexit = True # ensure callback invoked before exiting
>>> exit()
kenny2
 
No finalizer callbacks are ever called after the functions registered
with the atexit module. This avoid problems with invoking callbacks
later on during interpreter shutdown.
'''
 
import sys
import weakref
import atexit
 
 
class finalize(object):
'''
Class for registering a callback to be called at garbage collection
'''
__slots__ = ('atexit', '_sortkey')
_registry = {}
_shutdown = False
_count = 0
 
def __init__(self, obj, func, *args, **kwds):
self.atexit = False
self._sortkey = finalize._count
finalize._count += 1
if obj is None:
wr = None
else:
wr = weakref.ref(obj, self._make_wrapper())
self._registry[self] = (wr, func, args, kwds)
 
def __call__(self):
'''Invoke func(*args, **kwds) unless previously invoked'''
tmp = self._registry.pop(self, None)
if tmp and not self._shutdown:
_, func, args, kwds = tmp
return func(*args, **kwds)
 
def _make_wrapper(self):
'''Return a wrapper for self which can be used as weakref callback'''
def wrapper(wr):
try:
self()
except:
sys.stderr.write('Ignored exception in finalizer - ')
_print_stripped_exc(2)
return wrapper
 
 
# Print current exception with some frames stripped from traceback
def _print_stripped_exc(stripcount):
t, v, tb = sys.exc_info()
for i in range(stripcount):
if tb is None:
break
tb = tb.tb_next
if hasattr(v, '__traceback__'):
v.__traceback__ = tb
sys.excepthook(t, v, tb)
 
 
# At shutdown invoke finalizers for which atexit is true
def _atexit_callback():
try:
L = [f for f in finalize._registry.keys() if f.atexit]
L.sort(key=lambda f : f._sortkey, reverse=True)
for f in L:
f._make_wrapper()(None)
finally:
# prevent any more finalizers from executing during shutdown
finalize._shutdown = True
finalize._registry.clear()
 
atexit.register(_atexit_callback)
 
 
if __name__ == '__main__':
class A:
pass
 
def foo():
bar()
 
def bar():
1/0
 
L = []
a = A()
 
finalize(a, L.append, "hello 1")
f2 = finalize(a, L.append, "hello 2")
f3 = finalize(a, L.append, "hello 3")
finalize(a, foo)
 
L.append("111")
f3() # => L.append("hello 3")
L.append("222")
f3() # nothing because previously called
L.append("333")
print("[[[You should see a traceback for an ignored ZeroDivisionError:]]]")
del a # => print traceback for ZeroDivisionError
# => L.append("hello 2")
# => L.append("hello 1")
L.append("444")
f2() # nothing because previously called
 
print(L)
assert L == [
'111',
'hello 3',
'222',
'333',
'hello 2',
'hello 1',
'444'
]

Please sign in to comment on this gist.

Something went wrong with that request. Please try again.