1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
|
# -*- 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)
|
