Python个人代码库－beanstalkd操作类

#!/usr/bin/python
# -*- coding:UTF-8 -*-
import beanstalkc
import pickle
import time

class BSQueueC(object):
def __init__(self, host, port=11300):
self.host = host
self.port = port
self.__conn = beanstalkc.Connection(host, port)

def __del__(self):
self.__conn.close()

def put(self, tube, body, priority=2 ** 31, delay=0, ttr=120):
self.__conn.use(tube)
return self.__conn.put(body, priority, delay, ttr)

def put_dict(self, tube, dic, priority=2 ** 31, delay=0, ttr=120):
self.__conn.use(tube)
return self.__conn.put(body=pickle.dumps(dic), priority, delay, ttr)

def reserve(self, tube, timeout=0):
for t in self.__conn.watching():
self.__conn.ignore(t)
self.__conn.watch(tube)
return self.__conn.reserve(timeout)

def clear(self, tube):
while 1:
job = self.reserve(tube, 1)
if job is None:
break
else:
job.delete()

def is_empty(self, tube):
dic_stats = self.__conn.stats_tube(tube)
ready_job = dict_stats['current-jobs-ready']
reserver_job = dict_stats['current-jobs-reserved']
if (ready_job == 0) and (reserver_job == 0):
return True
else:
return False

def clear(self, tube):
try:
while 1:
job = self.reserve(tube, 1)
if job is None:
break
else:
job.delete()
except Exception, e:
print e

def watch_tubes(self, lis_tubes):
for t in self.__conn.watching():
self.__conn.ignore(t)
for tube in lis_tubes:
self.__conn.watch(tube.strip())

def kick_all_jobs(self, lis_tubes):
for t in lis_tubes:
self.__conn.use(t.strip())
self.__conn.kick()

def reserve_from_tubes(self, timeout=120):
return self.__conn.reserve(timeout)