Geek asks and answers

Python s context manager iterator socket

I have an iterator that returns context managers.

I need a pythonic statement withthat emulates the behavior of several nested statements with, one for each context manager returned by the iterator.

We can say that I want a generalization of the function (deprecated) contextlib.nested.

+4
source share
3 answers

contextlib.nested has two main problems due to which it is out of date.

  • , __init__ __new__, , all with __exit__ .
  • . , , True __exit__, . nested RuntimeError . , , nested.

, * nested! , nested ( , with ), . "iter_nested". , .

:

def contexts():
    yield MyContext1()
    yield MyContext2()

with iter_nested(contexts()) as contexts:
    do_stuff(contexts[0])
    do_other_stuff(contexts[1])

nested :

from contextlib import contextmanager

@contextmanager
--- def nested(*managers):
+++ def iter_nested(mgr_iterator):
    --- #comments & deprecation warning
    exits = []
    vars = []
    --- exc = (None, None, None)
    +++ exc = None # Python 3
    try:
        --- for mgr in managers:
        +++ for mgr in mgr_iterator:
            exit = mgr.__exit__
            enter = mgr.__enter__
            vars.append(enter())
            exits.append(exit)
        yield vars
# All of the following is new and fit for Python 3
except Exception as exception:
    exc = exception
    exc_tuple = (type(exc), exc, exc.__traceback__)
else:
    exc_tuple = (None, None, None)
finally:
    while exits:
        exit = exits.pop()
        try:
            if exit(*exc_tuple):
                exc = None
                exc_tuple = (None, None, None)
        except Exception as exception:
            exception.__context__ = exc
            exc = exception
            exc_tuple = (type(exc), exc, exc.__traceback__)
    if exc:
        raise exc
+1

docs:

, , warnings DeprecationWarning, [contextlib.nested], .

, : , __enter__ __enter__ . - , nested. , , . PEP-0343 .

+3

, - , , contextçlib.nested, , .

Contexts can be passed to it either as an object of a context protocol, or as a tuple, where the first member is the called object, which will be called with the remainder of the tuple as parameters in a managed environment:

import sys
import traceback


class NestContext(object):
    def __init__(self, *objects):
        self.objects = objects
    def __enter__(self):
        self.contexts = []
        for obj in self.objects:
            if isinstance(obj, tuple):
                try:
                    obj = obj[0](*obj[1:])
                except Exception, error:
                    self.__exit__(type(error), error, sys.exc_info()[2])
                    raise
            try:
                context = obj.__enter__()
            except Exception, error:
                self.__exit__(type(error), error, sys.exc_info()[2])
                raise   
            self.contexts.append(context)
        return self

    def __iter__(self):
        for context in self.contexts:
            yield context

    def __exit__(self, *args):
        for context in reversed(self.contexts):
            try:
                context.__exit__(*args)
            except Exception, error:
                sys.stderr.write(str(error))

if __name__ == "__main__":
    # example uasage

    class PlainContext(object):
        counter  = 0
        def __enter__(self):
            self.counter = self.__class__.counter
            print self.counter
            self.__class__.counter += 1
            return self
        def __exit__(self, *args):
            print "exiting %d" % self.counter

    with NestContext(*((PlainContext,) for i in range(10))) as all_contexts:
        print tuple(all_contexts)
0
source

More articles:


All Articles