Inspired by unittest.TestCase.addCleanup(), CleanupManager provides a means to programmatically add resources to be cleaned up when leaving a with statement. This makes it easy to use with optional resources, and those derived from sequences of inputs.
An more powerful version of this recipe with a few additional features is published under the name
ContextStack as part of the contextlib2 module: http://contextlib2.readthedocs.org
This recipe is based on a suggestion originally posted by Nikolaus Rath at http://bugs.python.org/issue13585
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
import collections import sys class CleanupManager(object): """Programmatic management of resource cleanup""" def __init__(self): self._callbacks = collections.deque() def register_exit(self, exit): """Accepts callbacks with the same signature as context manager __exit__ methods Can also suppress exceptions the same way __exit__ methods can. """ self._callbacks.append(exit) return exit # Allow use as a decorator def register(self, _cb, *args, **kwds): """Accepts arbitrary callbacks and arguments. Cannot suppress exceptions.""" def _wrapper(exc_type, exc, tb): _cb(*args, **kwds) return self.register_exit(_wrapper) def enter_context(self, cm): """Accepts and automatically enters other context managers""" # We look up the special methods on the type to match the with statement _cm_type = type(cm) _exit = _cm_type.__exit__ result = _cm_type.__enter__(cm) def _exit_wrapper(*exc_details): return _exit(cm, *exc_details) self.register_exit(_exit_wrapper) return result def close(self): self.__exit__(None, None, None) def __enter__(self): return self def __exit__(self, *exc_details): if not self._callbacks: return # This looks complicated, but it is really just # setting up a chain of try-expect statements to ensure # that outer callbacks still get invoked even if an # inner one throws an exception def _invoke_next_callback(exc_details): # Callbacks are removed from the list in FIFO order # but the recursion means they're *invoked* in LIFO order cb = self._callbacks.popleft() if not self._callbacks: # Innermost callback is invoked directly return cb(*exc_details) try: inner_result = _invoke_next_callback(exc_details) except: cb_result = cb(*sys.exc_info()) # Check if this cb suppressed the inner exception if not cb_result: raise else: # Check if inner cb suppressed the original exception if inner_result: exc_details = (None, None, None) cb_result = cb(*exc_details) or inner_result return cb_result # Kick off the recursive chain return _invoke_next_callback(exc_details)
A toy example that illustrates the LIFO execution of callbacks (necessary since callbacks registered later may depend on earlier resources):
>>> with CleanupManager() as cm: ... @cm.register # Registered first => cleaned up last ... def _exit(): ... print ("Hello world!") ... class TestCM(object): ... def __enter__(self): ... print ("Entered!") ... def __exit__(self, *exc_details): ... print ("Exited!") ... cm.enter_context(TestCM()) ... Entered! Exited! Hello world!
Cleaning up a list of files:
with contextlib.CleanupManager() as cm: files = [cm.enter_context(open(fname)) for fname in names] # All files will be closed when we leave the context
Nice. Tho, I'm curious about your thoughts on handling benign [__exit__] errors on cleanup.
I think any case could be handled by CM composition of some kind, but I'm currently imagining that being a little irritating.. Perhaps better stated, I'm thinking about cases where the caller wants the innermost result in the situation of outer result exceptions. (transient error filter?)
However, that is out of scope if we are strictly thinking about nested() done right.
Context manager exit methods in general don't have any information other than whether they received an exception or not. They never know where that exception came from (i.e. the contained block or some other context manager's exit method).
Basically, there's no such thing as a benign __exit__ error: they either return a false value (saying if there was an exception, reraise it, otherwise continue as normal) or a true one (saying to continue as normal, even if there was an exception). If they throw an exception themselves, it's as if it was thrown from the body of the with statement.
CleanupManager doesn't change any of that, but it does give you a few more options for structuring your exit handlers - such as pushing a "suppress transient errors" CM as the first one on the stack, or creating two separate cleanup managers. (Hmm, I'm tempted to change the name of this recipe to ContextStack and update the API accordingly...)
Yeah, "ContextStack" is a better name. "CleanupManager" lends to something a bit more abstract, which is probably what got me thinking about superfluous resource finalization/release errors in the first place.