sources for producer.py [rev. 38799]
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
"""
py lib's basic logging/tracing functionality 
    EXPERIMENTAL EXPERIMENTAL EXPERIMENTAL (especially the dispatching) 
WARNING: this module is not allowed to contain any 'py' imports, 
         Instead, it is very self-contained and should not depend on 
         CPython/stdlib versions, either.  One reason for these 
         restrictions is that this module should be sendable
         via py.execnet across the network in an very early phase.  
"""
class Message(object): 
    def __init__(self, keywords, args): 
        self.keywords = keywords 
        self.args = args 
    def content(self): 
        return " ".join(map(str, self.args))
    def prefix(self): 
        return "[%s] " % (":".join(self.keywords))
    def __str__(self): 
        return self.prefix() + self.content() 
class Producer(object):
    """ Log producer API which sends messages to be logged
        to a 'consumer' object, which then prints them to stdout,
        stderr, files, etc.
    """
    
    Message = Message  # to allow later customization 
    keywords2consumer = {}
    def __init__(self, keywords): 
        if isinstance(keywords, str): 
            keywords = tuple(keywords.split())
        self.keywords = keywords
    def __repr__(self):
        return "<py.log.Producer %s>" % ":".join(self.keywords) 
    def __getattr__(self, name):
        if '_' in name: 
            raise AttributeError, name
        producer = self.__class__(self.keywords + (name,))
        setattr(self, name, producer)
        return producer 
    
    def __call__(self, *args):
        """ write a message to the appropriate consumer(s) """
        func = self.get_consumer(self.keywords)
        if func is not None: 
            func(self.Message(self.keywords, args))
   
    def get_consumer(self, keywords): 
        """ return a consumer matching keywords
        
            tries to find the most suitable consumer by walking, starting from
            the back, the list of keywords, the first consumer matching a
            keyword is returned (falling back to py.log.default)
        """
        for i in range(len(self.keywords), 0, -1): 
            try: 
                return self.keywords2consumer[self.keywords[:i]]
            except KeyError: 
                continue
        return self.keywords2consumer.get('default', default_consumer)
    def set_consumer(self, consumer): 
        """ register a consumer matching our own keywords """
        self.keywords2consumer[self.keywords] = consumer 
default = Producer('default')
def _getstate(): 
    return Producer.keywords2consumer.copy()
def _setstate(state): 
    Producer.keywords2consumer.clear()
    Producer.keywords2consumer.update(state) 
def default_consumer(msg): 
    """ the default consumer, prints the message to stdout (using 'print') """
    print str(msg) 
Producer.keywords2consumer['default'] = default_consumer