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
|
import collections
import logging
class Sink(object):
def info(self, message, exc=None):
raise NotImplementedError
def warning(self, message, exc=None):
raise NotImplementedError
def error(self, message, exc=None):
raise NotImplementedError
class NullSink(Sink):
def info(self, message, exc=None):
pass
def warning(self, message, exc=None):
pass
def error(self, message, exc=None):
pass
class LoggingSink(Sink):
_DEFAULT_LOGGER = logging.getLogger(__name__)
def __init__(self, name=None, logger=None):
self.logger = logger
if self.logger is None and name is not None:
self.logger = logging.getLogger(name)
if self.logger is None:
self.logger = self._DEFAULT_LOGGER
def info(self, message, exc=None):
self.logger.info(message)
def warning(self, message, exc=None):
self.logger.warning(message)
def error(self, message, exc=None):
self.logger.error(message)
Message = collections.namedtuple('Message', ['type', 'content', 'exc'])
class CollectingSink(Sink):
def __init__(self):
self.messages = list()
def info(self, message, exc=None):
self.messages.append(Message('info', message, exc))
def warning(self, message, exc=None):
self.messages.append(Message('warning', message, exc))
def error(self, message, exc=None):
self.messages.append(Message('error', message, exc))
@property
def infos(self):
return [m for m in self.messages if m.type == 'info']
@property
def warnings(self):
return [m for m in self.messages if m.type == 'warning']
@property
def errors(self):
return [m for m in self.messages if m.type == 'error']
|
