"AbstractContextManager", "AbstractAsyncContextManager",

"AsyncExitStack", "ContextDecorator", "ExitStack",

"redirect_stdout", "redirect_stderr", "suppress", "aclosing",

"""An abstract base class for context managers."""

__class_getitem__ = classmethod(GenericAlias)

def __enter__(self):

"""Return `self` upon entering the runtime context."""

return self

@abc.abstractmethod

def __exit__(self, exc_type, exc_value, traceback):

"""Raise any exception triggered within the runtime context."""

return None

@classmethod

def __subclasshook__(cls, C):

if cls is AbstractContextManager:

return _collections_abc._check_methods(C, "__enter__", "__exit__")

"""An abstract base class for asynchronous context managers."""

__class_getitem__ = classmethod(GenericAlias)

async def __aenter__(self):

"""Return `self` upon entering the runtime context."""

return self

@abc.abstractmethod

async def __aexit__(self, exc_type, exc_value, traceback):

"""Raise any exception triggered within the runtime context."""

return None

@classmethod

def __subclasshook__(cls, C):

if cls is AbstractAsyncContextManager:

return _collections_abc._check_methods(C, "__aenter__",

"__aexit__")

"A base class or mixin that enables context managers to work as decorators."

def _recreate_cm(self):

"""Return a recreated instance of self.

return self

def __call__(self, func):

@wraps(func)

def inner(*args, **kwds):

with self._recreate_cm():

return func(*args, **kwds)

"A base class or mixin that enables async context managers to work as decorators."

def _recreate_cm(self):

"""Return a recreated instance of self.

return self

def __call__(self, func):

@wraps(func)

async def inner(*args, **kwds):

async with self._recreate_cm():

return await func(*args, **kwds)

"""Shared functionality for @contextmanager and @asynccontextmanager."""

def __init__(self, func, args, kwds):

self.gen = func(*args, **kwds)

self.func, self.args, self.kwds = func, args, kwds

# Issue 19330: ensure context manager instances have good docstrings

doc = getattr(func, "__doc__", None)

if doc is None:

doc = type(self).__doc__

self.__doc__ = doc

# Unfortunately, this still doesn't provide good help output when

# inspecting the created context manager instances, since pydoc

# currently bypasses the instance docstring and shows the docstring

# for the class instead.

# See http://bugs.python.org/issue19404 for more details.

def _recreate_cm(self):

# _GCMB instances are one-shot context managers, so the

# CM must be recreated each time a decorated function is

# called

_GeneratorContextManagerBase,

AbstractContextManager,

ContextDecorator,

):

"""Helper for @contextmanager decorator."""

def __enter__(self):

# do not keep args and kwds alive unnecessarily

# they are only needed for recreation, which is not possible anymore

del self.args, self.kwds, self.func

try:

return next(self.gen)

except StopIteration:

raise RuntimeError("generator didn't yield") from None

def __exit__(self, typ, value, traceback):

if typ is None:

try:

next(self.gen)

except StopIteration:

return False

else:

raise RuntimeError("generator didn't stop")

else:

if value is None:

# Need to force instantiation so we can reliably

# tell if we get the same exception back

value = typ()

try:

self.gen.throw(typ, value, traceback)

except StopIteration as exc:

# Suppress StopIteration *unless* it's the same exception that

# was passed to throw(). This prevents a StopIteration

# raised inside the "with" statement from being suppressed.

return exc is not value

except RuntimeError as exc:

# Don't re-raise the passed in exception. (issue27122)

if exc is value:

exc.__traceback__ = traceback

return False

# Avoid suppressing if a StopIteration exception

# was passed to throw() and later wrapped into a RuntimeError

# (see PEP 479 for sync generators; async generators also

# have this behavior). But do this only if the exception wrapped

# by the RuntimeError is actually Stop(Async)Iteration (see

# issue29692).

if (

isinstance(value, StopIteration)

and exc.__cause__ is value

):

exc.__traceback__ = traceback

return False

raise

except BaseException as exc:

# only re-raise if it's *not* the exception that was

# passed to throw(), because __exit__() must not raise

# an exception unless __exit__() itself failed. But throw()

# has to raise the exception to signal propagation, so this

# fixes the impedance mismatch between the throw() protocol

# and the __exit__() protocol.

if exc is not value:

raise

exc.__traceback__ = traceback

return False

_GeneratorContextManagerBase,

AbstractAsyncContextManager,

AsyncContextDecorator,

):

"""Helper for @asynccontextmanager decorator."""

async def __aenter__(self):

# do not keep args and kwds alive unnecessarily

# they are only needed for recreation, which is not possible anymore

del self.args, self.kwds, self.func

try:

return await anext(self.gen)

except StopAsyncIteration:

raise RuntimeError("generator didn't yield") from None

async def __aexit__(self, typ, value, traceback):

if typ is None:

try:

await anext(self.gen)

except StopAsyncIteration:

return False

else:

raise RuntimeError("generator didn't stop")

else:

if value is None:

# Need to force instantiation so we can reliably

# tell if we get the same exception back

value = typ()

try:

await self.gen.athrow(typ, value, traceback)

except StopAsyncIteration as exc:

# Suppress StopIteration *unless* it's the same exception that

# was passed to throw(). This prevents a StopIteration

# raised inside the "with" statement from being suppressed.

return exc is not value

except RuntimeError as exc:

# Don't re-raise the passed in exception. (issue27122)

if exc is value:

return False

# Avoid suppressing if a Stop(Async)Iteration exception

# was passed to athrow() and later wrapped into a RuntimeError

# (see PEP 479 for sync generators; async generators also

# have this behavior). But do this only if the exception wrapped

# by the RuntimeError is actually Stop(Async)Iteration (see

# issue29692).

if (

isinstance(value, (StopIteration, StopAsyncIteration))

and exc.__cause__ is value

):

return False

raise

except BaseException as exc:

# only re-raise if it's *not* the exception that was

# passed to throw(), because __exit__() must not raise

# an exception unless __exit__() itself failed. But throw()

# has to raise the exception to signal propagation, so this

# fixes the impedance mismatch between the throw() protocol

# and the __exit__() protocol.

if exc is not value:

raise

return False

"""@contextmanager decorator.

@wraps(func)

def helper(*args, **kwds):

return _GeneratorContextManager(func, args, kwds)

"""@asynccontextmanager decorator.

@wraps(func)

def helper(*args, **kwds):

return _AsyncGeneratorContextManager(func, args, kwds)

"""Context to automatically close something at the end of a block.

def __init__(self, thing):

self.thing = thing

def __enter__(self):

return self.thing

def __exit__(self, *exc_info):

"""Async context manager for safely finalizing an asynchronously cleaned-up

def __init__(self, thing):

self.thing = thing

async def __aenter__(self):

return self.thing

async def __aexit__(self, *exc_info):

_stream = None

def __init__(self, new_target):

self._new_target = new_target

# We use a list of old targets to make this CM re-entrant

self._old_targets = []

def __enter__(self):

self._old_targets.append(getattr(sys, self._stream))

setattr(sys, self._stream, self._new_target)

return self._new_target

def __exit__(self, exctype, excinst, exctb):

"""Context manager for temporarily redirecting stdout to another file.

"""Context manager to suppress specified exceptions

def __init__(self, *exceptions):

self._exceptions = exceptions

def __enter__(self):

pass

def __exit__(self, exctype, excinst, exctb):

# Unlike isinstance and issubclass, CPython exception handling

# currently only looks at the concrete type hierarchy (ignoring

# the instance and subclass checking hooks). While Guido considers

# that a bug rather than a feature, it's a fairly hard one to fix

# due to various internal implementation details. suppress provides

# the simpler issubclass based semantics, rather than trying to

# exactly reproduce the limitations of the CPython interpreter.

#

# See http://bugs.python.org/issue12029 for more details

"""A base class for ExitStack and AsyncExitStack."""

@staticmethod

def _create_exit_wrapper(cm, cm_exit):

return MethodType(cm_exit, cm)

@staticmethod

def _create_cb_wrapper(callback, /, *args, **kwds):

def _exit_wrapper(exc_type, exc, tb):

callback(*args, **kwds)

return _exit_wrapper

def __init__(self):

self._exit_callbacks = deque()

def pop_all(self):

"""Preserve the context stack by transferring it to a new instance."""

new_stack = type(self)()

new_stack._exit_callbacks = self._exit_callbacks

self._exit_callbacks = deque()

return new_stack

def push(self, exit):

"""Registers a callback with the standard __exit__ method signature.

# We use an unbound method rather than a bound method to follow

# the standard lookup behaviour for special methods.

_cb_type = type(exit)

try:

exit_method = _cb_type.__exit__

except AttributeError:

# Not a context manager, so assume it's a callable.

self._push_exit_callback(exit)

else:

self._push_cm_exit(exit, exit_method)

return exit # Allow use as a decorator.

def enter_context(self, cm):

"""Enters the supplied context manager.

# We look up the special methods on the type to match the with

# statement.

cls = type(cm)

try:

_enter = cls.__enter__

_exit = cls.__exit__

except AttributeError:

raise TypeError(f"'{cls.__module__}.{cls.__qualname__}' object does "

f"not support the context manager protocol") from None

result = _enter(cm)

self._push_cm_exit(cm, _exit)

return result

def callback(self, callback, /, *args, **kwds):

"""Registers an arbitrary callback and arguments.

_exit_wrapper = self._create_cb_wrapper(callback, *args, **kwds)

# We changed the signature, so using @wraps is not appropriate, but

# setting __wrapped__ may still help with introspection.

_exit_wrapper.__wrapped__ = callback

self._push_exit_callback(_exit_wrapper)

return callback # Allow use as a decorator

def _push_cm_exit(self, cm, cm_exit):

"""Helper to correctly register callbacks to __exit__ methods."""

_exit_wrapper = self._create_exit_wrapper(cm, cm_exit)

self._push_exit_callback(_exit_wrapper, True)

def _push_exit_callback(self, callback, is_sync=True):

"""Context manager for dynamic management of a stack of exit callbacks.

def __enter__(self):

return self

def __exit__(self, *exc_details):

received_exc = exc_details[0] is not None

# We manipulate the exception state so it behaves as though

# we were actually nesting multiple with statements

frame_exc = sys.exc_info()[1]

def _fix_exception_context(new_exc, old_exc):

# Context may not be correct, so find the end of the chain

while 1:

exc_context = new_exc.__context__

if exc_context is None or exc_context is old_exc:

# Context is already set correctly (see issue 20317)

return

if exc_context is frame_exc:

break

new_exc = exc_context

# Change the end of the chain to point to the exception

# we expect it to reference

new_exc.__context__ = old_exc

# Callbacks are invoked in LIFO order to match the behaviour of

# nested context managers

suppressed_exc = False

pending_raise = False

while self._exit_callbacks:

is_sync, cb = self._exit_callbacks.pop()

assert is_sync

try:

if cb(*exc_details):

suppressed_exc = True

pending_raise = False

exc_details = (None, None, None)

except:

new_exc_details = sys.exc_info()

# simulate the stack of exceptions by setting the context

_fix_exception_context(new_exc_details[1], exc_details[1])

pending_raise = True

exc_details = new_exc_details

if pending_raise:

try:

# bare "raise exc_details[1]" replaces our carefully

# set-up context

fixed_ctx = exc_details[1].__context__

raise exc_details[1]

except BaseException:

exc_details[1].__context__ = fixed_ctx

raise

return received_exc and suppressed_exc

def close(self):

"""Immediately unwind the context stack."""

"""Async context manager for dynamic management of a stack of exit

@staticmethod

def _create_async_exit_wrapper(cm, cm_exit):

return MethodType(cm_exit, cm)

@staticmethod

def _create_async_cb_wrapper(callback, /, *args, **kwds):

async def _exit_wrapper(exc_type, exc, tb):

await callback(*args, **kwds)

return _exit_wrapper

async def enter_async_context(self, cm):

"""Enters the supplied async context manager.

cls = type(cm)

try:

_enter = cls.__aenter__

_exit = cls.__aexit__

except AttributeError:

raise TypeError(f"'{cls.__module__}.{cls.__qualname__}' object does "

f"not support the asynchronous context manager protocol"

) from None

result = await _enter(cm)

self._push_async_cm_exit(cm, _exit)

return result

def push_async_exit(self, exit):

"""Registers a coroutine function with the standard __aexit__ method

_cb_type = type(exit)

try:

exit_method = _cb_type.__aexit__

except AttributeError:

# Not an async context manager, so assume it's a coroutine function

self._push_exit_callback(exit, False)

else:

self._push_async_cm_exit(exit, exit_method)

return exit # Allow use as a decorator

def push_async_callback(self, callback, /, *args, **kwds):

"""Registers an arbitrary coroutine function and arguments.

_exit_wrapper = self._create_async_cb_wrapper(callback, *args, **kwds)

# We changed the signature, so using @wraps is not appropriate, but

# setting __wrapped__ may still help with introspection.

_exit_wrapper.__wrapped__ = callback

self._push_exit_callback(_exit_wrapper, False)

return callback # Allow use as a decorator

async def aclose(self):

"""Immediately unwind the context stack."""

await self.__aexit__(None, None, None)

def _push_async_cm_exit(self, cm, cm_exit):

"""Helper to correctly register coroutine function to __aexit__

_exit_wrapper = self._create_async_exit_wrapper(cm, cm_exit)

self._push_exit_callback(_exit_wrapper, False)

async def __aenter__(self):

return self

async def __aexit__(self, *exc_details):

received_exc = exc_details[0] is not None

# We manipulate the exception state so it behaves as though

# we were actually nesting multiple with statements

frame_exc = sys.exc_info()[1]

def _fix_exception_context(new_exc, old_exc):

# Context may not be correct, so find the end of the chain

while 1:

exc_context = new_exc.__context__

if exc_context is None or exc_context is old_exc:

# Context is already set correctly (see issue 20317)

return

if exc_context is frame_exc:

break

new_exc = exc_context

# Change the end of the chain to point to the exception

# we expect it to reference

new_exc.__context__ = old_exc

# Callbacks are invoked in LIFO order to match the behaviour of

# nested context managers

suppressed_exc = False

pending_raise = False

while self._exit_callbacks:

is_sync, cb = self._exit_callbacks.pop()

try:

if is_sync:

cb_suppress = cb(*exc_details)

else:

cb_suppress = await cb(*exc_details)

if cb_suppress:

suppressed_exc = True

pending_raise = False

exc_details = (None, None, None)

except:

new_exc_details = sys.exc_info()

# simulate the stack of exceptions by setting the context

_fix_exception_context(new_exc_details[1], exc_details[1])

pending_raise = True

exc_details = new_exc_details

if pending_raise:

try:

# bare "raise exc_details[1]" replaces our carefully

# set-up context

fixed_ctx = exc_details[1].__context__

raise exc_details[1]

except BaseException:

exc_details[1].__context__ = fixed_ctx

raise

"""Context manager that does no additional processing.

def __init__(self, enter_result=None):

self.enter_result = enter_result

def __enter__(self):

return self.enter_result

def __exit__(self, *excinfo):

pass

async def __aenter__(self):

return self.enter_result

async def __aexit__(self, *excinfo):

"""Non thread-safe context manager to change the current working directory."""

def __init__(self, path):

self.path = path

self._old_cwd = []

def __enter__(self):

self._old_cwd.append(os.getcwd())

os.chdir(self.path)

def __exit__(self, *excinfo):