Class for finalization of weakrefable objects finalize(obj, func, *args, **kwargs) returns a callable finalizer object which will be called when obj is garbage collected. The first time the finalizer is called it evaluates func(*arg, **kwargs) and returns the result. After this the
| 440 | |
| 441 | |
| 442 | class finalize: |
| 443 | """Class for finalization of weakrefable objects |
| 444 | |
| 445 | finalize(obj, func, *args, **kwargs) returns a callable finalizer |
| 446 | object which will be called when obj is garbage collected. The |
| 447 | first time the finalizer is called it evaluates func(*arg, **kwargs) |
| 448 | and returns the result. After this the finalizer is dead, and |
| 449 | calling it just returns None. |
| 450 | |
| 451 | When the program exits any remaining finalizers for which the |
| 452 | atexit attribute is true will be run in reverse order of creation. |
| 453 | By default atexit is true. |
| 454 | """ |
| 455 | |
| 456 | # Finalizer objects don't have any state of their own. They are |
| 457 | # just used as keys to lookup _Info objects in the registry. This |
| 458 | # ensures that they cannot be part of a ref-cycle. |
| 459 | |
| 460 | __slots__ = () |
| 461 | _registry = {} |
| 462 | _shutdown = False |
| 463 | _index_iter = itertools.count() |
| 464 | _dirty = False |
| 465 | _registered_with_atexit = False |
| 466 | |
| 467 | class _Info: |
| 468 | __slots__ = ("weakref", "func", "args", "kwargs", "atexit", "index") |
| 469 | |
| 470 | def __init__(self, obj, func, /, *args, **kwargs): |
| 471 | if not self._registered_with_atexit: |
| 472 | # We may register the exit function more than once because |
| 473 | # of a thread race, but that is harmless |
| 474 | import atexit |
| 475 | atexit.register(self._exitfunc) |
| 476 | finalize._registered_with_atexit = True |
| 477 | info = self._Info() |
| 478 | info.weakref = ref(obj, self) |
| 479 | info.func = func |
| 480 | info.args = args |
| 481 | info.kwargs = kwargs or None |
| 482 | info.atexit = True |
| 483 | info.index = next(self._index_iter) |
| 484 | self._registry[self] = info |
| 485 | finalize._dirty = True |
| 486 | |
| 487 | def __call__(self, _=None): |
| 488 | """If alive then mark as dead and return func(*args, **kwargs); |
| 489 | otherwise return None""" |
| 490 | info = self._registry.pop(self, None) |
| 491 | if info and not self._shutdown: |
| 492 | return info.func(*info.args, **(info.kwargs or {})) |
| 493 | |
| 494 | def detach(self): |
| 495 | """If alive then mark as dead and return (obj, func, args, kwargs); |
| 496 | otherwise return None""" |
| 497 | info = self._registry.get(self) |
| 498 | obj = info and info.weakref() |
| 499 | if obj is not None and self._registry.pop(self, None): |
nothing calls this directly
no test coverage detected
searching dependent graphs…