"AbstractContextManager", "AbstractAsyncContextManager",

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

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

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

__class_getitem__ = classmethod(GenericAlias)

__slots__ = ()

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)

__slots__ = ()

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):

wrapper = wraps(func)

if _isasyncgenfunction(func):

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

with self._recreate_cm():

async with aclosing(func(*args, **kwds)) as gen:

async for value in gen:

yield value

return wrapper(asyncgen_inner)

elif _iscoroutinefunction(func):

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

with self._recreate_cm():

return await func(*args, **kwds)

return wrapper(async_inner)

elif _isgeneratorfunction(func):

def gen_inner(*args, **kwds):

with self._recreate_cm(), closing(func(*args, **kwds)) as gen:

return (yield from gen)

return wrapper(gen_inner)

else:

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):

wrapper = wraps(func)

if _isasyncgenfunction(func):

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

async with (

self._recreate_cm(),

aclosing(func(*args, **kwds)) as gen

):

async for value in gen:

yield value

return wrapper(asyncgen_inner)

elif _iscoroutinefunction(func):

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

async with self._recreate_cm():

return await func(*args, **kwds)

return wrapper(async_inner)

elif _isgeneratorfunction(func):

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

async with self._recreate_cm():

with closing(func(*args, **kwds)) as gen:

for value in gen:

yield value

return wrapper(gen_inner)

else:

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

async with self._recreate_cm():

return 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

# Slightly faster way to return next(self.gen):

for once in self.gen:

return once

raise RuntimeError("generator didn't yield")

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

if typ is None:

# Faster way to run next(self.gen) and check for StopIteration:

for _ in self.gen:

try:

raise RuntimeError("generator didn't stop")

finally:

self.gen.close()

return False

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(value)

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

):

value.__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

try:

raise RuntimeError("generator didn't stop after throw()")

finally:

_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:

try:

raise RuntimeError("generator didn't stop")

finally:

await self.gen.aclose()

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(value)

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:

exc.__traceback__ = traceback

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

):

value.__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

try:

raise RuntimeError("generator didn't stop after athrow()")

finally:

"""@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

if exctype is None:

return

if issubclass(exctype, self._exceptions):

return True

if issubclass(exctype, BaseExceptionGroup):

match, rest = excinst.split(self._exceptions)

if rest is None:

return True

raise rest

# Follow the standard lookup behaviour for special methods.

from inspect import getattr_static, _descriptor_get

cls = type(obj)

try:

descr = getattr_static(cls, name)

except AttributeError:

return default

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

@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.

exit_method = _lookup_special(exit, '__exit__', exit)

self._push_exit_callback(exit_method)

return exit # Allow use as a decorator.

def enter_context(self, cm):

"""Enters the supplied context manager.

_enter = _lookup_special(cm, '__enter__', _sentinel)

if _enter is _sentinel:

cls = type(cm)

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

f"not support the context manager protocol")

_exit = _lookup_special(cm, '__exit__', _sentinel)

if _exit is _sentinel:

cls = type(cm)

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

f"not support the context manager protocol")

result = _enter()

self._push_exit_callback(_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_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):

exc = exc_details[1]

received_exc = exc is not None

# We manipulate the exception state so it behaves as though

# we were actually nesting multiple with statements

frame_exc = sys.exception()

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 exc is None:

exc_details = None, None, None

else:

exc_details = type(exc), exc, exc.__traceback__

if cb(*exc_details):

suppressed_exc = True

pending_raise = False

exc = None

except BaseException as new_exc:

# simulate the stack of exceptions by setting the context

_fix_exception_context(new_exc, exc)

pending_raise = True

exc = new_exc

if pending_raise:

try:

# bare "raise exc" replaces our carefully

# set-up context

fixed_ctx = exc.__context__

raise exc

except BaseException:

exc.__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_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.

_enter = _lookup_special(cm, '__aenter__', _sentinel)

if _enter is _sentinel:

cls = type(cm)

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

f"not support the asynchronous context manager protocol")

_exit = _lookup_special(cm, '__aexit__', _sentinel)

if _exit is _sentinel:

cls = type(cm)

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

f"not support the asynchronous context manager protocol")

result = await _enter()

self._push_exit_callback(_exit, False)

return result

def push_async_exit(self, exit):

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

exit_method = _lookup_special(exit, '__aexit__', exit)

self._push_exit_callback(exit_method, False)

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)

async def __aenter__(self):

return self

async def __aexit__(self, *exc_details):

exc = exc_details[1]

received_exc = exc is not None

# We manipulate the exception state so it behaves as though

# we were actually nesting multiple with statements

frame_exc = sys.exception()

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 exc is None:

exc_details = None, None, None

else:

exc_details = type(exc), exc, exc.__traceback__

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 = None

except BaseException as new_exc:

# simulate the stack of exceptions by setting the context

_fix_exception_context(new_exc, exc)

pending_raise = True

exc = new_exc

if pending_raise:

try:

# bare "raise exc" replaces our carefully

# set-up context

fixed_ctx = exc.__context__

raise exc

except BaseException:

exc.__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):