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
|
from __future__ import absolute_import, print_function, division
import contextlib
import blinker
import pprint
from mitmproxy import exceptions
class Options(object):
"""
.changed is a blinker Signal that triggers whenever options are
updated. If any handler in the chain raises an exceptions.OptionsError
exception, all changes are rolled back, the exception is suppressed,
and the .errored signal is notified.
"""
_initialized = False
attributes = []
def __new__(cls, *args, **kwargs):
# Initialize instance._opts before __init__ is called.
# This allows us to call super().__init__() last, which then sets
# ._initialized = True as the final operation.
instance = super(Options, cls).__new__(cls)
instance.__dict__["_opts"] = {}
return instance
def __init__(self):
self.__dict__["changed"] = blinker.Signal()
self.__dict__["errored"] = blinker.Signal()
self.__dict__["_initialized"] = True
@contextlib.contextmanager
def rollback(self):
old = self._opts.copy()
try:
yield
except exceptions.OptionsError as e:
# Notify error handlers
self.errored.send(self, exc=e)
# Rollback
self.__dict__["_opts"] = old
self.changed.send(self)
def __eq__(self, other):
return self._opts == other._opts
def __copy__(self):
return self.__class__(**self._opts)
def __getattr__(self, attr):
if attr in self._opts:
return self._opts[attr]
else:
raise AttributeError()
def __setattr__(self, attr, value):
if not self._initialized:
self._opts[attr] = value
return
if attr not in self._opts:
raise KeyError("No such option: %s" % attr)
with self.rollback():
self._opts[attr] = value
self.changed.send(self)
def get(self, k, d=None):
return self._opts.get(k, d)
def update(self, **kwargs):
for k in kwargs:
if k not in self._opts:
raise KeyError("No such option: %s" % k)
with self.rollback():
self._opts.update(kwargs)
self.changed.send(self)
def setter(self, attr):
if attr not in self._opts:
raise KeyError("No such option: %s" % attr)
return lambda x: self.__setattr__(attr, x)
def __repr__(self):
options = pprint.pformat(self._opts, indent=4).strip(" {}")
if "\n" in options:
options = "\n " + options + "\n"
return "{mod}.{cls}({{{options}}})".format(
mod=type(self).__module__,
cls=type(self).__name__,
options=options
)
|
