1 """Utilities for with-statement contexts. See PEP 343."""
2 import abc
3 import os
4 import sys
5 import _collections_abc
6 from collections import deque
7 from functools import wraps
8 from types import MethodType, GenericAlias
9
10 __all__ = ["asynccontextmanager", "contextmanager", "closing", "nullcontext",
11 "AbstractContextManager", "AbstractAsyncContextManager",
12 "AsyncExitStack", "ContextDecorator", "ExitStack",
13 "redirect_stdout", "redirect_stderr", "suppress", "aclosing",
14 "chdir"]
15
16
17 class ESC[4;38;5;81mAbstractContextManager(ESC[4;38;5;149mabcESC[4;38;5;149m.ESC[4;38;5;149mABC):
18
19 """An abstract base class for context managers."""
20
21 __class_getitem__ = classmethod(GenericAlias)
22
23 def __enter__(self):
24 """Return `self` upon entering the runtime context."""
25 return self
26
27 @abc.abstractmethod
28 def __exit__(self, exc_type, exc_value, traceback):
29 """Raise any exception triggered within the runtime context."""
30 return None
31
32 @classmethod
33 def __subclasshook__(cls, C):
34 if cls is AbstractContextManager:
35 return _collections_abc._check_methods(C, "__enter__", "__exit__")
36 return NotImplemented
37
38
39 class ESC[4;38;5;81mAbstractAsyncContextManager(ESC[4;38;5;149mabcESC[4;38;5;149m.ESC[4;38;5;149mABC):
40
41 """An abstract base class for asynchronous context managers."""
42
43 __class_getitem__ = classmethod(GenericAlias)
44
45 async def __aenter__(self):
46 """Return `self` upon entering the runtime context."""
47 return self
48
49 @abc.abstractmethod
50 async def __aexit__(self, exc_type, exc_value, traceback):
51 """Raise any exception triggered within the runtime context."""
52 return None
53
54 @classmethod
55 def __subclasshook__(cls, C):
56 if cls is AbstractAsyncContextManager:
57 return _collections_abc._check_methods(C, "__aenter__",
58 "__aexit__")
59 return NotImplemented
60
61
62 class ESC[4;38;5;81mContextDecorator(ESC[4;38;5;149mobject):
63 "A base class or mixin that enables context managers to work as decorators."
64
65 def _recreate_cm(self):
66 """Return a recreated instance of self.
67
68 Allows an otherwise one-shot context manager like
69 _GeneratorContextManager to support use as
70 a decorator via implicit recreation.
71
72 This is a private interface just for _GeneratorContextManager.
73 See issue #11647 for details.
74 """
75 return self
76
77 def __call__(self, func):
78 @wraps(func)
79 def inner(*args, **kwds):
80 with self._recreate_cm():
81 return func(*args, **kwds)
82 return inner
83
84
85 class ESC[4;38;5;81mAsyncContextDecorator(ESC[4;38;5;149mobject):
86 "A base class or mixin that enables async context managers to work as decorators."
87
88 def _recreate_cm(self):
89 """Return a recreated instance of self.
90 """
91 return self
92
93 def __call__(self, func):
94 @wraps(func)
95 async def inner(*args, **kwds):
96 async with self._recreate_cm():
97 return await func(*args, **kwds)
98 return inner
99
100
101 class ESC[4;38;5;81m_GeneratorContextManagerBase:
102 """Shared functionality for @contextmanager and @asynccontextmanager."""
103
104 def __init__(self, func, args, kwds):
105 self.gen = func(*args, **kwds)
106 self.func, self.args, self.kwds = func, args, kwds
107 # Issue 19330: ensure context manager instances have good docstrings
108 doc = getattr(func, "__doc__", None)
109 if doc is None:
110 doc = type(self).__doc__
111 self.__doc__ = doc
112 # Unfortunately, this still doesn't provide good help output when
113 # inspecting the created context manager instances, since pydoc
114 # currently bypasses the instance docstring and shows the docstring
115 # for the class instead.
116 # See http://bugs.python.org/issue19404 for more details.
117
118 def _recreate_cm(self):
119 # _GCMB instances are one-shot context managers, so the
120 # CM must be recreated each time a decorated function is
121 # called
122 return self.__class__(self.func, self.args, self.kwds)
123
124
125 class ESC[4;38;5;81m_GeneratorContextManager(
126 ESC[4;38;5;149m_GeneratorContextManagerBase,
127 ESC[4;38;5;149mAbstractContextManager,
128 ESC[4;38;5;149mContextDecorator,
129 ):
130 """Helper for @contextmanager decorator."""
131
132 def __enter__(self):
133 # do not keep args and kwds alive unnecessarily
134 # they are only needed for recreation, which is not possible anymore
135 del self.args, self.kwds, self.func
136 try:
137 return next(self.gen)
138 except StopIteration:
139 raise RuntimeError("generator didn't yield") from None
140
141 def __exit__(self, typ, value, traceback):
142 if typ is None:
143 try:
144 next(self.gen)
145 except StopIteration:
146 return False
147 else:
148 try:
149 raise RuntimeError("generator didn't stop")
150 finally:
151 self.gen.close()
152 else:
153 if value is None:
154 # Need to force instantiation so we can reliably
155 # tell if we get the same exception back
156 value = typ()
157 try:
158 self.gen.throw(typ, value, traceback)
159 except StopIteration as exc:
160 # Suppress StopIteration *unless* it's the same exception that
161 # was passed to throw(). This prevents a StopIteration
162 # raised inside the "with" statement from being suppressed.
163 return exc is not value
164 except RuntimeError as exc:
165 # Don't re-raise the passed in exception. (issue27122)
166 if exc is value:
167 exc.__traceback__ = traceback
168 return False
169 # Avoid suppressing if a StopIteration exception
170 # was passed to throw() and later wrapped into a RuntimeError
171 # (see PEP 479 for sync generators; async generators also
172 # have this behavior). But do this only if the exception wrapped
173 # by the RuntimeError is actually Stop(Async)Iteration (see
174 # issue29692).
175 if (
176 isinstance(value, StopIteration)
177 and exc.__cause__ is value
178 ):
179 value.__traceback__ = traceback
180 return False
181 raise
182 except BaseException as exc:
183 # only re-raise if it's *not* the exception that was
184 # passed to throw(), because __exit__() must not raise
185 # an exception unless __exit__() itself failed. But throw()
186 # has to raise the exception to signal propagation, so this
187 # fixes the impedance mismatch between the throw() protocol
188 # and the __exit__() protocol.
189 if exc is not value:
190 raise
191 exc.__traceback__ = traceback
192 return False
193 try:
194 raise RuntimeError("generator didn't stop after throw()")
195 finally:
196 self.gen.close()
197
198 class ESC[4;38;5;81m_AsyncGeneratorContextManager(
199 ESC[4;38;5;149m_GeneratorContextManagerBase,
200 ESC[4;38;5;149mAbstractAsyncContextManager,
201 ESC[4;38;5;149mAsyncContextDecorator,
202 ):
203 """Helper for @asynccontextmanager decorator."""
204
205 async def __aenter__(self):
206 # do not keep args and kwds alive unnecessarily
207 # they are only needed for recreation, which is not possible anymore
208 del self.args, self.kwds, self.func
209 try:
210 return await anext(self.gen)
211 except StopAsyncIteration:
212 raise RuntimeError("generator didn't yield") from None
213
214 async def __aexit__(self, typ, value, traceback):
215 if typ is None:
216 try:
217 await anext(self.gen)
218 except StopAsyncIteration:
219 return False
220 else:
221 try:
222 raise RuntimeError("generator didn't stop")
223 finally:
224 await self.gen.aclose()
225 else:
226 if value is None:
227 # Need to force instantiation so we can reliably
228 # tell if we get the same exception back
229 value = typ()
230 try:
231 await self.gen.athrow(typ, value, traceback)
232 except StopAsyncIteration as exc:
233 # Suppress StopIteration *unless* it's the same exception that
234 # was passed to throw(). This prevents a StopIteration
235 # raised inside the "with" statement from being suppressed.
236 return exc is not value
237 except RuntimeError as exc:
238 # Don't re-raise the passed in exception. (issue27122)
239 if exc is value:
240 exc.__traceback__ = traceback
241 return False
242 # Avoid suppressing if a Stop(Async)Iteration exception
243 # was passed to athrow() and later wrapped into a RuntimeError
244 # (see PEP 479 for sync generators; async generators also
245 # have this behavior). But do this only if the exception wrapped
246 # by the RuntimeError is actually Stop(Async)Iteration (see
247 # issue29692).
248 if (
249 isinstance(value, (StopIteration, StopAsyncIteration))
250 and exc.__cause__ is value
251 ):
252 value.__traceback__ = traceback
253 return False
254 raise
255 except BaseException as exc:
256 # only re-raise if it's *not* the exception that was
257 # passed to throw(), because __exit__() must not raise
258 # an exception unless __exit__() itself failed. But throw()
259 # has to raise the exception to signal propagation, so this
260 # fixes the impedance mismatch between the throw() protocol
261 # and the __exit__() protocol.
262 if exc is not value:
263 raise
264 exc.__traceback__ = traceback
265 return False
266 try:
267 raise RuntimeError("generator didn't stop after athrow()")
268 finally:
269 await self.gen.aclose()
270
271
272 def contextmanager(func):
273 """@contextmanager decorator.
274
275 Typical usage:
276
277 @contextmanager
278 def some_generator(<arguments>):
279 <setup>
280 try:
281 yield <value>
282 finally:
283 <cleanup>
284
285 This makes this:
286
287 with some_generator(<arguments>) as <variable>:
288 <body>
289
290 equivalent to this:
291
292 <setup>
293 try:
294 <variable> = <value>
295 <body>
296 finally:
297 <cleanup>
298 """
299 @wraps(func)
300 def helper(*args, **kwds):
301 return _GeneratorContextManager(func, args, kwds)
302 return helper
303
304
305 def asynccontextmanager(func):
306 """@asynccontextmanager decorator.
307
308 Typical usage:
309
310 @asynccontextmanager
311 async def some_async_generator(<arguments>):
312 <setup>
313 try:
314 yield <value>
315 finally:
316 <cleanup>
317
318 This makes this:
319
320 async with some_async_generator(<arguments>) as <variable>:
321 <body>
322
323 equivalent to this:
324
325 <setup>
326 try:
327 <variable> = <value>
328 <body>
329 finally:
330 <cleanup>
331 """
332 @wraps(func)
333 def helper(*args, **kwds):
334 return _AsyncGeneratorContextManager(func, args, kwds)
335 return helper
336
337
338 class ESC[4;38;5;81mclosing(ESC[4;38;5;149mAbstractContextManager):
339 """Context to automatically close something at the end of a block.
340
341 Code like this:
342
343 with closing(<module>.open(<arguments>)) as f:
344 <block>
345
346 is equivalent to this:
347
348 f = <module>.open(<arguments>)
349 try:
350 <block>
351 finally:
352 f.close()
353
354 """
355 def __init__(self, thing):
356 self.thing = thing
357 def __enter__(self):
358 return self.thing
359 def __exit__(self, *exc_info):
360 self.thing.close()
361
362
363 class ESC[4;38;5;81maclosing(ESC[4;38;5;149mAbstractAsyncContextManager):
364 """Async context manager for safely finalizing an asynchronously cleaned-up
365 resource such as an async generator, calling its ``aclose()`` method.
366
367 Code like this:
368
369 async with aclosing(<module>.fetch(<arguments>)) as agen:
370 <block>
371
372 is equivalent to this:
373
374 agen = <module>.fetch(<arguments>)
375 try:
376 <block>
377 finally:
378 await agen.aclose()
379
380 """
381 def __init__(self, thing):
382 self.thing = thing
383 async def __aenter__(self):
384 return self.thing
385 async def __aexit__(self, *exc_info):
386 await self.thing.aclose()
387
388
389 class ESC[4;38;5;81m_RedirectStream(ESC[4;38;5;149mAbstractContextManager):
390
391 _stream = None
392
393 def __init__(self, new_target):
394 self._new_target = new_target
395 # We use a list of old targets to make this CM re-entrant
396 self._old_targets = []
397
398 def __enter__(self):
399 self._old_targets.append(getattr(sys, self._stream))
400 setattr(sys, self._stream, self._new_target)
401 return self._new_target
402
403 def __exit__(self, exctype, excinst, exctb):
404 setattr(sys, self._stream, self._old_targets.pop())
405
406
407 class ESC[4;38;5;81mredirect_stdout(ESC[4;38;5;149m_RedirectStream):
408 """Context manager for temporarily redirecting stdout to another file.
409
410 # How to send help() to stderr
411 with redirect_stdout(sys.stderr):
412 help(dir)
413
414 # How to write help() to a file
415 with open('help.txt', 'w') as f:
416 with redirect_stdout(f):
417 help(pow)
418 """
419
420 _stream = "stdout"
421
422
423 class ESC[4;38;5;81mredirect_stderr(ESC[4;38;5;149m_RedirectStream):
424 """Context manager for temporarily redirecting stderr to another file."""
425
426 _stream = "stderr"
427
428
429 class ESC[4;38;5;81msuppress(ESC[4;38;5;149mAbstractContextManager):
430 """Context manager to suppress specified exceptions
431
432 After the exception is suppressed, execution proceeds with the next
433 statement following the with statement.
434
435 with suppress(FileNotFoundError):
436 os.remove(somefile)
437 # Execution still resumes here if the file was already removed
438 """
439
440 def __init__(self, *exceptions):
441 self._exceptions = exceptions
442
443 def __enter__(self):
444 pass
445
446 def __exit__(self, exctype, excinst, exctb):
447 # Unlike isinstance and issubclass, CPython exception handling
448 # currently only looks at the concrete type hierarchy (ignoring
449 # the instance and subclass checking hooks). While Guido considers
450 # that a bug rather than a feature, it's a fairly hard one to fix
451 # due to various internal implementation details. suppress provides
452 # the simpler issubclass based semantics, rather than trying to
453 # exactly reproduce the limitations of the CPython interpreter.
454 #
455 # See http://bugs.python.org/issue12029 for more details
456 return exctype is not None and issubclass(exctype, self._exceptions)
457
458
459 class ESC[4;38;5;81m_BaseExitStack:
460 """A base class for ExitStack and AsyncExitStack."""
461
462 @staticmethod
463 def _create_exit_wrapper(cm, cm_exit):
464 return MethodType(cm_exit, cm)
465
466 @staticmethod
467 def _create_cb_wrapper(callback, /, *args, **kwds):
468 def _exit_wrapper(exc_type, exc, tb):
469 callback(*args, **kwds)
470 return _exit_wrapper
471
472 def __init__(self):
473 self._exit_callbacks = deque()
474
475 def pop_all(self):
476 """Preserve the context stack by transferring it to a new instance."""
477 new_stack = type(self)()
478 new_stack._exit_callbacks = self._exit_callbacks
479 self._exit_callbacks = deque()
480 return new_stack
481
482 def push(self, exit):
483 """Registers a callback with the standard __exit__ method signature.
484
485 Can suppress exceptions the same way __exit__ method can.
486 Also accepts any object with an __exit__ method (registering a call
487 to the method instead of the object itself).
488 """
489 # We use an unbound method rather than a bound method to follow
490 # the standard lookup behaviour for special methods.
491 _cb_type = type(exit)
492
493 try:
494 exit_method = _cb_type.__exit__
495 except AttributeError:
496 # Not a context manager, so assume it's a callable.
497 self._push_exit_callback(exit)
498 else:
499 self._push_cm_exit(exit, exit_method)
500 return exit # Allow use as a decorator.
501
502 def enter_context(self, cm):
503 """Enters the supplied context manager.
504
505 If successful, also pushes its __exit__ method as a callback and
506 returns the result of the __enter__ method.
507 """
508 # We look up the special methods on the type to match the with
509 # statement.
510 cls = type(cm)
511 try:
512 _enter = cls.__enter__
513 _exit = cls.__exit__
514 except AttributeError:
515 raise TypeError(f"'{cls.__module__}.{cls.__qualname__}' object does "
516 f"not support the context manager protocol") from None
517 result = _enter(cm)
518 self._push_cm_exit(cm, _exit)
519 return result
520
521 def callback(self, callback, /, *args, **kwds):
522 """Registers an arbitrary callback and arguments.
523
524 Cannot suppress exceptions.
525 """
526 _exit_wrapper = self._create_cb_wrapper(callback, *args, **kwds)
527
528 # We changed the signature, so using @wraps is not appropriate, but
529 # setting __wrapped__ may still help with introspection.
530 _exit_wrapper.__wrapped__ = callback
531 self._push_exit_callback(_exit_wrapper)
532 return callback # Allow use as a decorator
533
534 def _push_cm_exit(self, cm, cm_exit):
535 """Helper to correctly register callbacks to __exit__ methods."""
536 _exit_wrapper = self._create_exit_wrapper(cm, cm_exit)
537 self._push_exit_callback(_exit_wrapper, True)
538
539 def _push_exit_callback(self, callback, is_sync=True):
540 self._exit_callbacks.append((is_sync, callback))
541
542
543 # Inspired by discussions on http://bugs.python.org/issue13585
544 class ESC[4;38;5;81mExitStack(ESC[4;38;5;149m_BaseExitStack, ESC[4;38;5;149mAbstractContextManager):
545 """Context manager for dynamic management of a stack of exit callbacks.
546
547 For example:
548 with ExitStack() as stack:
549 files = [stack.enter_context(open(fname)) for fname in filenames]
550 # All opened files will automatically be closed at the end of
551 # the with statement, even if attempts to open files later
552 # in the list raise an exception.
553 """
554
555 def __enter__(self):
556 return self
557
558 def __exit__(self, *exc_details):
559 received_exc = exc_details[0] is not None
560
561 # We manipulate the exception state so it behaves as though
562 # we were actually nesting multiple with statements
563 frame_exc = sys.exc_info()[1]
564 def _fix_exception_context(new_exc, old_exc):
565 # Context may not be correct, so find the end of the chain
566 while 1:
567 exc_context = new_exc.__context__
568 if exc_context is None or exc_context is old_exc:
569 # Context is already set correctly (see issue 20317)
570 return
571 if exc_context is frame_exc:
572 break
573 new_exc = exc_context
574 # Change the end of the chain to point to the exception
575 # we expect it to reference
576 new_exc.__context__ = old_exc
577
578 # Callbacks are invoked in LIFO order to match the behaviour of
579 # nested context managers
580 suppressed_exc = False
581 pending_raise = False
582 while self._exit_callbacks:
583 is_sync, cb = self._exit_callbacks.pop()
584 assert is_sync
585 try:
586 if cb(*exc_details):
587 suppressed_exc = True
588 pending_raise = False
589 exc_details = (None, None, None)
590 except:
591 new_exc_details = sys.exc_info()
592 # simulate the stack of exceptions by setting the context
593 _fix_exception_context(new_exc_details[1], exc_details[1])
594 pending_raise = True
595 exc_details = new_exc_details
596 if pending_raise:
597 try:
598 # bare "raise exc_details[1]" replaces our carefully
599 # set-up context
600 fixed_ctx = exc_details[1].__context__
601 raise exc_details[1]
602 except BaseException:
603 exc_details[1].__context__ = fixed_ctx
604 raise
605 return received_exc and suppressed_exc
606
607 def close(self):
608 """Immediately unwind the context stack."""
609 self.__exit__(None, None, None)
610
611
612 # Inspired by discussions on https://bugs.python.org/issue29302
613 class ESC[4;38;5;81mAsyncExitStack(ESC[4;38;5;149m_BaseExitStack, ESC[4;38;5;149mAbstractAsyncContextManager):
614 """Async context manager for dynamic management of a stack of exit
615 callbacks.
616
617 For example:
618 async with AsyncExitStack() as stack:
619 connections = [await stack.enter_async_context(get_connection())
620 for i in range(5)]
621 # All opened connections will automatically be released at the
622 # end of the async with statement, even if attempts to open a
623 # connection later in the list raise an exception.
624 """
625
626 @staticmethod
627 def _create_async_exit_wrapper(cm, cm_exit):
628 return MethodType(cm_exit, cm)
629
630 @staticmethod
631 def _create_async_cb_wrapper(callback, /, *args, **kwds):
632 async def _exit_wrapper(exc_type, exc, tb):
633 await callback(*args, **kwds)
634 return _exit_wrapper
635
636 async def enter_async_context(self, cm):
637 """Enters the supplied async context manager.
638
639 If successful, also pushes its __aexit__ method as a callback and
640 returns the result of the __aenter__ method.
641 """
642 cls = type(cm)
643 try:
644 _enter = cls.__aenter__
645 _exit = cls.__aexit__
646 except AttributeError:
647 raise TypeError(f"'{cls.__module__}.{cls.__qualname__}' object does "
648 f"not support the asynchronous context manager protocol"
649 ) from None
650 result = await _enter(cm)
651 self._push_async_cm_exit(cm, _exit)
652 return result
653
654 def push_async_exit(self, exit):
655 """Registers a coroutine function with the standard __aexit__ method
656 signature.
657
658 Can suppress exceptions the same way __aexit__ method can.
659 Also accepts any object with an __aexit__ method (registering a call
660 to the method instead of the object itself).
661 """
662 _cb_type = type(exit)
663 try:
664 exit_method = _cb_type.__aexit__
665 except AttributeError:
666 # Not an async context manager, so assume it's a coroutine function
667 self._push_exit_callback(exit, False)
668 else:
669 self._push_async_cm_exit(exit, exit_method)
670 return exit # Allow use as a decorator
671
672 def push_async_callback(self, callback, /, *args, **kwds):
673 """Registers an arbitrary coroutine function and arguments.
674
675 Cannot suppress exceptions.
676 """
677 _exit_wrapper = self._create_async_cb_wrapper(callback, *args, **kwds)
678
679 # We changed the signature, so using @wraps is not appropriate, but
680 # setting __wrapped__ may still help with introspection.
681 _exit_wrapper.__wrapped__ = callback
682 self._push_exit_callback(_exit_wrapper, False)
683 return callback # Allow use as a decorator
684
685 async def aclose(self):
686 """Immediately unwind the context stack."""
687 await self.__aexit__(None, None, None)
688
689 def _push_async_cm_exit(self, cm, cm_exit):
690 """Helper to correctly register coroutine function to __aexit__
691 method."""
692 _exit_wrapper = self._create_async_exit_wrapper(cm, cm_exit)
693 self._push_exit_callback(_exit_wrapper, False)
694
695 async def __aenter__(self):
696 return self
697
698 async def __aexit__(self, *exc_details):
699 received_exc = exc_details[0] is not None
700
701 # We manipulate the exception state so it behaves as though
702 # we were actually nesting multiple with statements
703 frame_exc = sys.exc_info()[1]
704 def _fix_exception_context(new_exc, old_exc):
705 # Context may not be correct, so find the end of the chain
706 while 1:
707 exc_context = new_exc.__context__
708 if exc_context is None or exc_context is old_exc:
709 # Context is already set correctly (see issue 20317)
710 return
711 if exc_context is frame_exc:
712 break
713 new_exc = exc_context
714 # Change the end of the chain to point to the exception
715 # we expect it to reference
716 new_exc.__context__ = old_exc
717
718 # Callbacks are invoked in LIFO order to match the behaviour of
719 # nested context managers
720 suppressed_exc = False
721 pending_raise = False
722 while self._exit_callbacks:
723 is_sync, cb = self._exit_callbacks.pop()
724 try:
725 if is_sync:
726 cb_suppress = cb(*exc_details)
727 else:
728 cb_suppress = await cb(*exc_details)
729
730 if cb_suppress:
731 suppressed_exc = True
732 pending_raise = False
733 exc_details = (None, None, None)
734 except:
735 new_exc_details = sys.exc_info()
736 # simulate the stack of exceptions by setting the context
737 _fix_exception_context(new_exc_details[1], exc_details[1])
738 pending_raise = True
739 exc_details = new_exc_details
740 if pending_raise:
741 try:
742 # bare "raise exc_details[1]" replaces our carefully
743 # set-up context
744 fixed_ctx = exc_details[1].__context__
745 raise exc_details[1]
746 except BaseException:
747 exc_details[1].__context__ = fixed_ctx
748 raise
749 return received_exc and suppressed_exc
750
751
752 class ESC[4;38;5;81mnullcontext(ESC[4;38;5;149mAbstractContextManager, ESC[4;38;5;149mAbstractAsyncContextManager):
753 """Context manager that does no additional processing.
754
755 Used as a stand-in for a normal context manager, when a particular
756 block of code is only sometimes used with a normal context manager:
757
758 cm = optional_cm if condition else nullcontext()
759 with cm:
760 # Perform operation, using optional_cm if condition is True
761 """
762
763 def __init__(self, enter_result=None):
764 self.enter_result = enter_result
765
766 def __enter__(self):
767 return self.enter_result
768
769 def __exit__(self, *excinfo):
770 pass
771
772 async def __aenter__(self):
773 return self.enter_result
774
775 async def __aexit__(self, *excinfo):
776 pass
777
778
779 class ESC[4;38;5;81mchdir(ESC[4;38;5;149mAbstractContextManager):
780 """Non thread-safe context manager to change the current working directory."""
781
782 def __init__(self, path):
783 self.path = path
784 self._old_cwd = []
785
786 def __enter__(self):
787 self._old_cwd.append(os.getcwd())
788 os.chdir(self.path)
789
790 def __exit__(self, *excinfo):
791 os.chdir(self._old_cwd.pop())