aboutsummaryrefslogtreecommitdiff
path: root/cros_utils/contextlib3.py
diff options
context:
space:
mode:
Diffstat (limited to 'cros_utils/contextlib3.py')
-rw-r--r--cros_utils/contextlib3.py116
1 files changed, 0 insertions, 116 deletions
diff --git a/cros_utils/contextlib3.py b/cros_utils/contextlib3.py
deleted file mode 100644
index 9fabbf6e..00000000
--- a/cros_utils/contextlib3.py
+++ /dev/null
@@ -1,116 +0,0 @@
-# -*- coding: utf-8 -*-
-# Copyright 2019 The Chromium OS Authors. All rights reserved.
-# Use of this source code is governed by a BSD-style license that can be
-# found in the LICENSE file.
-
-"""Random utilties from Python3's contextlib."""
-
-from __future__ import division
-from __future__ import print_function
-
-import sys
-
-
-class ExitStack(object):
- """https://docs.python.org/3/library/contextlib.html#contextlib.ExitStack"""
-
- def __init__(self):
- self._stack = []
- self._is_entered = False
-
- def _assert_is_entered(self):
- # Strictly, entering has no effect on the operations that call this.
- # However, if you're trying to e.g. push things to an ExitStack that hasn't
- # yet been entered, that's likely a bug.
- assert self._is_entered, 'ExitStack op performed before entering'
-
- def __enter__(self):
- self._is_entered = True
- return self
-
- def _perform_exit(self, exc_type, exc, exc_traceback):
- # I suppose a better name for this is
- # `take_exception_handling_into_our_own_hands`, but that's harder to type.
- exception_handled = False
- while self._stack:
- fn = self._stack.pop()
- # The except clause below is meant to run as-if it's a `finally` block,
- # but `finally` blocks don't have easy access to exceptions currently in
- # flight. Hence, we do need to catch things like KeyboardInterrupt,
- # SystemExit, ...
- # pylint: disable=bare-except
- try:
- # If an __exit__ handler returns a truthy value, we should assume that
- # it handled the exception appropriately. Otherwise, we need to keep it
- # with us. (PEP 343)
- if fn(exc_type, exc, exc_traceback):
- exc_type, exc, exc_traceback = None, None, None
- exception_handled = True
- except:
- # Python2 doesn't appear to have the notion of 'exception causes',
- # which is super unfortunate. In the case:
- #
- # @contextlib.contextmanager
- # def foo()
- # try:
- # yield
- # finally:
- # raise ValueError
- #
- # with foo():
- # assert False
- #
- # ...Python will only note the ValueError; nothing about the failing
- # assertion is printed.
- #
- # I guess on the bright side, that means we don't have to fiddle with
- # __cause__s/etc.
- exc_type, exc, exc_traceback = sys.exc_info()
- exception_handled = True
-
- if not exception_handled:
- return False
-
- # Something changed. We either need to raise for ourselves, or note that
- # the exception has been suppressed.
- if exc_type is not None:
- raise exc_type, exc, exc_traceback
-
- # Otherwise, the exception was suppressed. Go us!
- return True
-
- def __exit__(self, exc_type, exc, exc_traceback):
- return self._perform_exit(exc_type, exc, exc_traceback)
-
- def close(self):
- """Unwinds the exit stack, unregistering all events"""
- self._perform_exit(None, None, None)
-
- def enter_context(self, cm):
- """Enters the given context manager, and registers it to be exited."""
- self._assert_is_entered()
-
- # The spec specifically notes that we should take __exit__ prior to calling
- # __enter__.
- exit_cleanup = cm.__exit__
- result = cm.__enter__()
- self._stack.append(exit_cleanup)
- return result
-
- # pylint complains about `exit` being redefined. `exit` is the documented
- # name of this param, and renaming it would break portability if someone
- # decided to `push(exit=foo)`, so just ignore the lint.
- # pylint: disable=redefined-builtin
- def push(self, exit):
- """Like `enter_context`, but won't enter the value given."""
- self._assert_is_entered()
- self._stack.append(exit.__exit__)
-
- def callback(self, callback, *args, **kwargs):
- """Performs the given callback on exit"""
- self._assert_is_entered()
-
- def fn(_exc_type, _exc, _exc_traceback):
- callback(*args, **kwargs)
-
- self._stack.append(fn)