def finalize(self):
"""Finalizes the object if not already done.
Returns: None
"""
# this is the "public" finalize method
raise NotImplementedError(
"finalize() must be implemented by AutoFinalizedObject."
)
python类finalize()的实例源码
def __del__(self):
self.finalize()
def _do_finalize_object_ref(obj_ref):
"""Helper function for weakref.finalize() that dereferences a weakref
to an object and calls its _do_finalize_object() method if the object
is still alive. Does nothing otherwise.
Returns: None (implicit)
Arguments:
* obj_ref -- weakref to an object
"""
obj = obj_ref()
if obj is not None:
# else object disappeared
obj._do_finalize_object()
def __new__(cls, *args, **kwargs):
"""Creates a new object instance and adds the private finalizer
attributes to it.
Returns: new object instance
Arguments:
* *args, **kwargs -- passed to the parent instance creator
(which ignores them)
"""
# Note: Do not pass a (hard) reference to instance to the
# finalizer as func/args/kwargs, it'd keep the object
# alive until the program terminates.
# A weak reference is fine.
#
# Note 2: When using weakrefs and not calling finalize() in
# __del__, the object may already have disappeared
# when weakref.finalize() kicks in.
# Make sure that _finalizer() gets called,
# i.e. keep __del__() from the base class.
#
# Note 3: the _finalize_called attribute is (probably) useless
# for this class
instance = super(AutoFinalizedObject, cls).__new__(
cls, *args, **kwargs
)
instance._finalizer = weakref.finalize(
instance, _do_finalize_object_ref, weakref.ref(instance)
)
return instance
def finalize(self):
"""Finalizes the object if not already done."""
self._finalizer()
def test_finalize(self):
def add(x,y,z):
res.append(x + y + z)
return x + y + z
a = self.A()
res = []
f = weakref.finalize(a, add, 67, 43, z=89)
self.assertEqual(f.alive, True)
self.assertEqual(f.peek(), (a, add, (67,43), {'z':89}))
self.assertEqual(f(), 199)
self.assertEqual(f(), None)
self.assertEqual(f(), None)
self.assertEqual(f.peek(), None)
self.assertEqual(f.detach(), None)
self.assertEqual(f.alive, False)
self.assertEqual(res, [199])
res = []
f = weakref.finalize(a, add, 67, 43, 89)
self.assertEqual(f.peek(), (a, add, (67,43,89), {}))
self.assertEqual(f.detach(), (a, add, (67,43,89), {}))
self.assertEqual(f(), None)
self.assertEqual(f(), None)
self.assertEqual(f.peek(), None)
self.assertEqual(f.detach(), None)
self.assertEqual(f.alive, False)
self.assertEqual(res, [])
res = []
f = weakref.finalize(a, add, x=67, y=43, z=89)
del a
self._collect_if_necessary()
self.assertEqual(f(), None)
self.assertEqual(f(), None)
self.assertEqual(f.peek(), None)
self.assertEqual(f.detach(), None)
self.assertEqual(f.alive, False)
self.assertEqual(res, [199])
def test_order(self):
a = self.A()
res = []
f1 = weakref.finalize(a, res.append, 'f1')
f2 = weakref.finalize(a, res.append, 'f2')
f3 = weakref.finalize(a, res.append, 'f3')
f4 = weakref.finalize(a, res.append, 'f4')
f5 = weakref.finalize(a, res.append, 'f5')
# make sure finalizers can keep themselves alive
del f1, f4
self.assertTrue(f2.alive)
self.assertTrue(f3.alive)
self.assertTrue(f5.alive)
self.assertTrue(f5.detach())
self.assertFalse(f5.alive)
f5() # nothing because previously unregistered
res.append('A')
f3() # => res.append('f3')
self.assertFalse(f3.alive)
res.append('B')
f3() # nothing because previously called
res.append('C')
del a
self._collect_if_necessary()
# => res.append('f4')
# => res.append('f2')
# => res.append('f1')
self.assertFalse(f2.alive)
res.append('D')
f2() # nothing because previously called by gc
expected = ['A', 'f3', 'B', 'C', 'f4', 'f2', 'f1', 'D']
self.assertEqual(res, expected)
def __init__(self, suffix="", prefix=template, dir=None):
self.name = mkdtemp(suffix, prefix, dir)
self._finalizer = _weakref.finalize(
self, self._cleanup, self.name,
warn_message="Implicitly cleaning up {!r}".format(self))
def __init__(self, suffix=None, prefix=None, dir=None):
self.name = mkdtemp(suffix, prefix, dir)
self._finalizer = _weakref.finalize(
self, self._cleanup, self.name,
warn_message="Implicitly cleaning up {!r}".format(self))
def __init__(self, suffix=None, prefix=None, dir=None):
self.name = mkdtemp(suffix, prefix, dir)
self._finalizer = _weakref.finalize(
self, self._cleanup, self.name,
warn_message="Implicitly cleaning up {!r}".format(self))
def __init__(self, suffix=None, prefix=None, dir=None):
self.name = mkdtemp(suffix, prefix, dir)
self._finalizer = _weakref.finalize(
self, self._cleanup, self.name,
warn_message="Implicitly cleaning up {!r}".format(self))
def __init__(self, suffix=None, prefix=None, dir=None):
self.name = mkdtemp(suffix, prefix, dir)
self._finalizer = _weakref.finalize(
self, self._cleanup, self.name,
warn_message="Implicitly cleaning up {!r}".format(self))
def test_finalize(self):
def add(x,y,z):
res.append(x + y + z)
return x + y + z
a = self.A()
res = []
f = weakref.finalize(a, add, 67, 43, z=89)
self.assertEqual(f.alive, True)
self.assertEqual(f.peek(), (a, add, (67,43), {'z':89}))
self.assertEqual(f(), 199)
self.assertEqual(f(), None)
self.assertEqual(f(), None)
self.assertEqual(f.peek(), None)
self.assertEqual(f.detach(), None)
self.assertEqual(f.alive, False)
self.assertEqual(res, [199])
res = []
f = weakref.finalize(a, add, 67, 43, 89)
self.assertEqual(f.peek(), (a, add, (67,43,89), {}))
self.assertEqual(f.detach(), (a, add, (67,43,89), {}))
self.assertEqual(f(), None)
self.assertEqual(f(), None)
self.assertEqual(f.peek(), None)
self.assertEqual(f.detach(), None)
self.assertEqual(f.alive, False)
self.assertEqual(res, [])
res = []
f = weakref.finalize(a, add, x=67, y=43, z=89)
del a
self._collect_if_necessary()
self.assertEqual(f(), None)
self.assertEqual(f(), None)
self.assertEqual(f.peek(), None)
self.assertEqual(f.detach(), None)
self.assertEqual(f.alive, False)
self.assertEqual(res, [199])
def test_order(self):
a = self.A()
res = []
f1 = weakref.finalize(a, res.append, 'f1')
f2 = weakref.finalize(a, res.append, 'f2')
f3 = weakref.finalize(a, res.append, 'f3')
f4 = weakref.finalize(a, res.append, 'f4')
f5 = weakref.finalize(a, res.append, 'f5')
# make sure finalizers can keep themselves alive
del f1, f4
self.assertTrue(f2.alive)
self.assertTrue(f3.alive)
self.assertTrue(f5.alive)
self.assertTrue(f5.detach())
self.assertFalse(f5.alive)
f5() # nothing because previously unregistered
res.append('A')
f3() # => res.append('f3')
self.assertFalse(f3.alive)
res.append('B')
f3() # nothing because previously called
res.append('C')
del a
self._collect_if_necessary()
# => res.append('f4')
# => res.append('f2')
# => res.append('f1')
self.assertFalse(f2.alive)
res.append('D')
f2() # nothing because previously called by gc
expected = ['A', 'f3', 'B', 'C', 'f4', 'f2', 'f1', 'D']
self.assertEqual(res, expected)
def __init__(self, suffix="", prefix=template, dir=None):
self.name = mkdtemp(suffix, prefix, dir)
self._finalizer = _weakref.finalize(
self, self._cleanup, self.name,
warn_message="Implicitly cleaning up {!r}".format(self))
def connect(self, receiver, **kwargs):
obj = receiver
receiver = weakref.ref(receiver)
weakref.finalize(obj, self.flag)
self.receivers.append(receiver)
self.clear()
def finalize(self):
"""Finalizes the object if not already done.
Returns: None
"""
# this is the "public" finalize method
raise NotImplementedError(
"finalize() must be implemented by AutoFinalizedObject."
)
def __del__(self):
self.finalize()
def _do_finalize_object_ref(obj_ref):
"""Helper function for weakref.finalize() that dereferences a weakref
to an object and calls its _do_finalize_object() method if the object
is still alive. Does nothing otherwise.
Returns: None (implicit)
Arguments:
* obj_ref -- weakref to an object
"""
obj = obj_ref()
if obj is not None:
# else object disappeared
obj._do_finalize_object()
def __new__(cls, *args, **kwargs):
"""Creates a new object instance and adds the private finalizer
attributes to it.
Returns: new object instance
Arguments:
* *args, **kwargs -- passed to the parent instance creator
(which ignores them)
"""
# Note: Do not pass a (hard) reference to instance to the
# finalizer as func/args/kwargs, it'd keep the object
# alive until the program terminates.
# A weak reference is fine.
#
# Note 2: When using weakrefs and not calling finalize() in
# __del__, the object may already have disappeared
# when weakref.finalize() kicks in.
# Make sure that _finalizer() gets called,
# i.e. keep __del__() from the base class.
#
# Note 3: the _finalize_called attribute is (probably) useless
# for this class
instance = super(AutoFinalizedObject, cls).__new__(
cls, *args, **kwargs
)
instance._finalizer = weakref.finalize(
instance, _do_finalize_object_ref, weakref.ref(instance)
)
return instance
