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
|
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.
"""
attributes = []
def __init__(self, **kwargs):
self.__dict__["changed"] = blinker.Signal()
self.__dict__["errored"] = blinker.Signal()
self.__dict__["_opts"] = dict([(i, None) for i in self.attributes])
for k, v in kwargs.items():
self._opts[k] = v
@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):
return self._opts[attr]
def __setattr__(self, attr, value):
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 __repr__(self):
return pprint.pformat(self._opts)
|
