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
|
#
# Copyright (C) 2011-2013 Vinay Sajip. See LICENSE.txt for details.
#
"""
This module contains classes which help you work with Redis queues.
"""
from logutils.queue import QueueHandler, QueueListener
try:
import cPickle as pickle
except ImportError:
import pickle
class RedisQueueHandler(QueueHandler):
"""
A QueueHandler implementation which pushes pickled
records to a Redis queue using a specified key.
:param key: The key to use for the queue. Defaults to
"python.logging".
:param redis: If specified, this instance is used to
communicate with a Redis instance.
:param limit: If specified, the queue is restricted to
have only this many elements.
"""
def __init__(self, key='python.logging', redis=None, limit=0):
if redis is None:
from redis import Redis
redis = Redis()
self.key = key
assert limit >= 0
self.limit = limit
QueueHandler.__init__(self, redis)
def enqueue(self, record):
s = pickle.dumps(vars(record))
self.queue.rpush(self.key, s)
if self.limit:
self.queue.ltrim(self.key, -self.limit, -1)
class RedisQueueListener(QueueListener):
"""
A QueueListener implementation which fetches pickled
records from a Redis queue using a specified key.
:param key: The key to use for the queue. Defaults to
"python.logging".
:param redis: If specified, this instance is used to
communicate with a Redis instance.
"""
def __init__(self, *handlers, **kwargs):
redis = kwargs.get('redis')
if redis is None:
from redis import Redis
redis = Redis()
self.key = kwargs.get('key', 'python.logging')
QueueListener.__init__(self, redis, *handlers)
def dequeue(self, block):
"""
Dequeue and return a record.
"""
if block:
s = self.queue.blpop(self.key)[1]
else:
s = self.queue.lpop(self.key)
if not s:
record = None
else:
record = pickle.loads(s)
return record
def enqueue_sentinel(self):
self.queue.rpush(self.key, '')
|
