python - Alternative to contextlib.nested with variable number of context managers -
we have code invokes variable number of context managers depending on runtime parameters:
from contextlib import nested, contextmanager @contextmanager def my_context(arg): print("entering", arg) try: yield arg finally: print("exiting", arg) def my_fn(items): nested(*(my_context(arg) arg in items)) managers: print("processing under", managers) my_fn(range(3))
however, contextlib.nested
deprecated since python 2.7:
deprecationwarning: with-statements directly support multiple context managers
the answers multiple variables in python 'with' statement indicate contextlib.nested
has "confusing error prone quirks", suggested alternative of using multiple-manager with
statement won't work variable number of context managers (and breaks backward compatibility).
are there alternatives contextlib.nested
aren't deprecated , (preferably) don't have same bugs?
or should continue use contextlib.nested
, ignore warning? if so, should plan contextlib.nested
removed @ time in future?
the new python 3 contextlib.exitstack
class added replacement contextlib.nested()
(see issue 13585).
it coded in such way can use in python 2 directly:
import sys collections import deque class exitstack(object): """context manager dynamic management of stack of exit callbacks example: exitstack() stack: files = [stack.enter_context(open(fname)) fname in filenames] # opened files automatically closed @ end of # statement, if attempts open files later # in list raise exception """ def __init__(self): self._exit_callbacks = deque() def pop_all(self): """preserve context stack transferring new instance""" new_stack = type(self)() new_stack._exit_callbacks = self._exit_callbacks self._exit_callbacks = deque() return new_stack def _push_cm_exit(self, cm, cm_exit): """helper correctly register callbacks __exit__ methods""" def _exit_wrapper(*exc_details): return cm_exit(cm, *exc_details) _exit_wrapper.__self__ = cm self.push(_exit_wrapper) def push(self, exit): """registers callback standard __exit__ method signature can suppress exceptions same way __exit__ methods can. accepts object __exit__ method (registering call method instead of object itself) """ # use unbound method rather bound method follow # standard lookup behaviour special methods _cb_type = type(exit) try: exit_method = _cb_type.__exit__ except attributeerror: # not context manager, assume callable self._exit_callbacks.append(exit) else: self._push_cm_exit(exit, exit_method) return exit # allow use decorator def callback(self, callback, *args, **kwds): """registers arbitrary callback , arguments. cannot suppress exceptions. """ def _exit_wrapper(exc_type, exc, tb): callback(*args, **kwds) # changed signature, using @wraps not appropriate, # setting __wrapped__ may still introspection _exit_wrapper.__wrapped__ = callback self.push(_exit_wrapper) return callback # allow use decorator def enter_context(self, cm): """enters supplied context manager if successful, pushes __exit__ method callback , returns result of __enter__ method. """ # special methods on type match statement _cm_type = type(cm) _exit = _cm_type.__exit__ result = _cm_type.__enter__(cm) self._push_cm_exit(cm, _exit) return result def close(self): """immediately unwind context stack""" self.__exit__(none, none, none) def __enter__(self): return self def __exit__(self, *exc_details): # manipulate exception state behaves though # nesting multiple statements frame_exc = sys.exc_info()[1] def _fix_exception_context(new_exc, old_exc): while 1: exc_context = new_exc.__context__ if exc_context in (none, frame_exc): break new_exc = exc_context new_exc.__context__ = old_exc # callbacks invoked in lifo order match behaviour of # nested context managers suppressed_exc = false while self._exit_callbacks: cb = self._exit_callbacks.pop() try: if cb(*exc_details): suppressed_exc = true exc_details = (none, none, none) except: new_exc_details = sys.exc_info() # simulate stack of exceptions setting context _fix_exception_context(new_exc_details[1], exc_details[1]) if not self._exit_callbacks: raise exc_details = new_exc_details return suppressed_exc
use context manager, add nested context managers @ will:
with exitstack() stack: managers = [stack.enter_context(my_context(arg)) arg in items] print("processing under", managers)
for example context manager, prints:
>>> my_fn(range(3)) ('entering', 0) ('entering', 1) ('entering', 2) ('processing under', [0, 1, 2]) ('exiting', 2) ('exiting', 1) ('exiting', 0)
you can install contextlib2
module; includes exitstack
backport.
Comments
Post a Comment