This issue tracker has been migrated to GitHub, and is currently read-only.
For more information, see the GitHub FAQs in the Python's Developer Guide.

Author ncoghlan
Recipients Julian, eric.snow, giampaolo.rodola, ncoghlan, nikratio, rhettinger, smarnach
Date 2011-12-13.02:11:41
SpamBayes Score 0.0
Marked as misclassified No
Message-id <1323742304.86.0.52511491969.issue13585@psf.upfronthosting.co.za>
In-reply-to
Content
Given the existence of tempfile.TemporaryDirectory in recent Python versions, I suggest finding a new cleanup function example that doesn't duplicate native stdlib functionality :)

I do see value in the feature itself though - I believe the precedent of both the atexit module [1] and unittest.TestCase.addCleanup() [2] shows how it can be useful to have a standard way to accumulate a sequence of "undo" operations for invocation at a later time.

In particular, it's a much better fit than nested with statements are for *optional* resources - you can make the with statement unconditional (setting up the cleanup manager), optionally add the cleanup methods, and avoid needing to have two copies of your actual invocation code (one inside a with statement and one without) or having a delayed check in a finally block.

It codifies the fairly common "if this resource was acquired, make sure it is released" idiom in a way that with statements and try/finally just don't handle neatly. By using an incremental API, it also avoids the traps associated with the ultimately misguided "contextlib.nested()" design.

However, I suggest using an API that strictly follows the "register only" model employed by both of the existing mechanisms. In addition, any solution provided as part of contextlib should interoperate nicely with existing context managers - it should be trivial to rewrite a nested with statement to be based on CleanupManager instead.

Accordingly, I would give the manager the following public methods:

  register_exit() (only accepts callbacks with the __exit__ signature)
  register() (equivalent to TestCase.addCleanup)
  enter_context() (accepts actual context managers)
  close() (equivalent to TestCase.doCleanups)

register_exit() would be the base callback registration method. It would accept only callbacks with the same signature as __exit__ methods.

    def register_exit(self, exit):
        self._callbacks.append(exit)
        return exit # Allow use as a decorator

register() would wrap arbitrary callbacks to support the __exit__ method signature:

    def register(self, _cb, *args, **kwds):
        def _wrapper(exc_type, exc, tb):
            return _cb(*args, **kwds)
        return self.register_exit(_wrapper)

enter_context() would work as follows:

    def enter_context(self, cm):
        result = cm.__enter__()
        self.register_exit(cm.__exit__)
        return result

close() would look like:

    def close(self):
        self.__exit__(None, None, None)

And finally, __exit__() itself would be:

    def __exit__(self, *exc_details):
        def _invoke_next_callback(exc_details):
            # Callbacks are removed from the list in FIFO order
            # but are actually *invoked* in LIFO order
            cb = self._callbacks.pop(0)
            if not self._callbacks:
                # Innermost callback is invoked directly
                return cb(exc_type, exc, tb)
            # Use try-finally to ensure this callback still gets
            # invoked even if an inner one fails
            try:
                inner_result = _invoke_next_callback()
            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)
            return cb_result
        _invoke_next_callback(exc_details)

An example using a cleanup manager to handle multiple files, one of which is optional:

    with contextlib.CleanupManager() as cm:
        source = cm.enter_context(open(source_fname))
        if dest_fname is not None:
            dest = cm.enter_context(open(dest_fname))
            _write_to_dest = dest.write
        else:
            def _write_to_dest(line): pass
        for line in source:
            _write_to_dest(line)
            yield line


[1] http://docs.python.org/library/atexit
[2] http://docs.python.org/library/unittest#unittest.TestCase.addCleanup
History
Date User Action Args
2011-12-13 02:11:45ncoghlansetrecipients: + ncoghlan, rhettinger, giampaolo.rodola, nikratio, Julian, eric.snow, smarnach
2011-12-13 02:11:44ncoghlansetmessageid: <1323742304.86.0.52511491969.issue13585@psf.upfronthosting.co.za>
2011-12-13 02:11:43ncoghlanlinkissue13585 messages
2011-12-13 02:11:41ncoghlancreate