<s> """<STR_LIT>""" <EOL> import redis <EOL> from redis . exceptions import ConnectionError <EOL> from functools import wraps <EOL> try : <EOL> from itertools import zip_longest <EOL> except ImportError : <EOL> from itertools import izip_longest as zip_longest <EOL> import logging <EOL> logger = logging . getLogger ( __name__ ) <EOL> class Job ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , id , queue_name , payload ) : <EOL> """<STR_LIT>""" <EOL> self . id = id <EOL> self . queue_name = queue_name <EOL> self . payload = payload <EOL> def __repr__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT>' % ( self . id , self . queue_name ) <EOL> class Node ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , node_id , host , port , connection ) : <EOL> """<STR_LIT>""" <EOL> self . node_id = node_id <EOL> self . host = host <EOL> self . port = port <EOL> self . connection = connection <EOL> def __repr__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return '<STR_LIT>' % ( self . host , self . port ) <EOL> class retry ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , retry_count = <NUM_LIT:2> ) : <EOL> """<STR_LIT>""" <EOL> self . retry_count = retry_count <EOL> def __call__ ( self , fn ) : <EOL> """<STR_LIT>""" <EOL> @ wraps ( fn ) <EOL> def wrapped_f ( * args , ** kwargs ) : <EOL> c = <NUM_LIT:0> <EOL> while c <= self . retry_count : <EOL> try : <EOL> return fn ( * args , ** kwargs ) <EOL> except : <EOL> logging . critical ( "<STR_LIT>" , <EOL> c ) <EOL> logging . exception ( "<STR_LIT>" ) <EOL> if c == self . retry_count : <EOL> raise <EOL> c += <NUM_LIT:1> <EOL> return wrapped_f <EOL> class Client ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , nodes = None ) : <EOL> """<STR_LIT>""" <EOL> if nodes is None : <EOL> nodes = [ '<STR_LIT>' ] <EOL> self . nodes = { } <EOL> for n in nodes : <EOL> self . nodes [ n ] = None <EOL> self . connected_node = None <EOL> def connect ( self ) : <EOL> """<STR_LIT>""" <EOL> self . connected_node = None <EOL> for i , node in self . nodes . items ( ) : <EOL> host , port = i . split ( '<STR_LIT::>' ) <EOL> port = int ( port ) <EOL> redis_client = redis . Redis ( host , port ) <EOL> try : <EOL> ret = redis_client . execute_command ( '<STR_LIT>' ) <EOL> format_version , node_id = ret [ <NUM_LIT:0> ] , ret [ <NUM_LIT:1> ] <EOL> others = ret [ <NUM_LIT:2> : ] <EOL> self . nodes [ i ] = Node ( node_id , host , port , redis_client ) <EOL> self . connected_node = self . nodes [ i ] <EOL> except redis . exceptions . ConnectionError : <EOL> pass <EOL> if not self . connected_node : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> logger . info ( "<STR_LIT>" % self . connected_node ) <EOL> def get_connection ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . connected_node . connection <EOL> @ retry ( ) <EOL> def execute_command ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return self . get_connection ( ) . execute_command ( * args , ** kwargs ) <EOL> except ConnectionError as e : <EOL> logger . warn ( '<STR_LIT>' ) <EOL> self . connect ( ) <EOL> logger . warn ( '<STR_LIT>' ) <EOL> raise <EOL> def _grouper ( self , iterable , n , fillvalue = None ) : <EOL> """<STR_LIT>""" <EOL> args = [ iter ( iterable ) ] * n <EOL> return zip_longest ( fillvalue = fillvalue , * args ) <EOL> def info ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . execute_command ( "<STR_LIT>" ) <EOL> def add_job ( self , queue_name , job , timeout = <NUM_LIT:200> , replicate = None , delay = None , <EOL> retry = None , ttl = None , maxlen = None , async = None ) : <EOL> """<STR_LIT>""" <EOL> command = [ '<STR_LIT>' , queue_name , job , timeout ] <EOL> if replicate : <EOL> command += [ '<STR_LIT>' , replicate ] <EOL> if delay : <EOL> command += [ '<STR_LIT>' , delay ] <EOL> if retry : <EOL> command += [ '<STR_LIT>' , retry ] <EOL> if ttl : <EOL> command += [ '<STR_LIT>' , ttl ] <EOL> if maxlen : <EOL> command += [ '<STR_LIT>' , maxlen ] <EOL> if async : <EOL> command += [ '<STR_LIT>' ] <EOL> logger . debug ( "<STR_LIT>" , command ) <EOL> job_id = self . execute_command ( * command ) <EOL> logger . debug ( "<STR_LIT>" , command ) <EOL> logger . debug ( "<STR_LIT>" % job_id ) <EOL> return job_id <EOL> def get_job ( self , queues , timeout = None , count = None , nohang = False , withcounters = False ) : <EOL> """<STR_LIT>""" <EOL> assert queues <EOL> command = [ '<STR_LIT>' ] <EOL> if nohang : <EOL> command += [ '<STR_LIT>' ] <EOL> if timeout : <EOL> command += [ '<STR_LIT>' , timeout ] <EOL> if count : <EOL> command += [ '<STR_LIT>' , count ] <EOL> if withcounters : <EOL> command += [ '<STR_LIT>' ] <EOL> command += [ '<STR_LIT>' ] + queues <EOL> results = self . execute_command ( * command ) <EOL> if not results : <EOL> return [ ] <EOL> if withcounters : <EOL> return [ ( job_id , queue_name , job , nacks , additional_deliveries ) for <EOL> job_id , queue_name , job , _ , nacks , _ , additional_deliveries in results ] <EOL> else : <EOL> return [ ( job_id , queue_name , job ) for <EOL> job_id , queue_name , job in results ] <EOL> def ack_job ( self , * job_ids ) : <EOL> """<STR_LIT>""" <EOL> self . execute_command ( '<STR_LIT>' , * job_ids ) <EOL> def nack_job ( self , * job_ids ) : <EOL> """<STR_LIT>""" <EOL> self . execute_command ( '<STR_LIT>' , * job_ids ) <EOL> def fast_ack ( self , * job_ids ) : <EOL> """<STR_LIT>""" <EOL> self . execute_command ( '<STR_LIT>' , * job_ids ) <EOL> def working ( self , job_id ) : <EOL> """<STR_LIT>""" <EOL> return self . execute_command ( '<STR_LIT>' , job_id ) <EOL> def qlen ( self , queue_name ) : <EOL> """<STR_LIT>""" <EOL> return self . execute_command ( '<STR_LIT>' , queue_name ) <EOL> def qstat ( self , queue_name , return_dict = False ) : <EOL> """<STR_LIT>""" <EOL> rtn = self . execute_command ( '<STR_LIT>' , queue_name ) <EOL> if return_dict : <EOL> grouped = self . _grouper ( rtn , <NUM_LIT:2> ) <EOL> rtn = dict ( ( a , b ) for a , b in grouped ) <EOL> return rtn <EOL> def qpeek ( self , queue_name , count ) : <EOL> """<STR_LIT>""" <EOL> return self . execute_command ( "<STR_LIT>" , queue_name , count ) <EOL> def enqueue ( self , * job_ids ) : <EOL> """<STR_LIT>""" <EOL> return self . execute_command ( "<STR_LIT>" , * job_ids ) <EOL> def dequeue ( self , * job_ids ) : <EOL> """<STR_LIT>""" <EOL> return self . execute_command ( "<STR_LIT>" , * job_ids ) <EOL> def del_job ( self , * job_ids ) : <EOL> """<STR_LIT>""" <EOL> return self . execute_command ( "<STR_LIT>" , * job_ids ) <EOL> def show ( self , job_id , return_dict = False ) : <EOL> """<STR_LIT>""" <EOL> rtn = self . execute_command ( '<STR_LIT>' , job_id ) <EOL> if return_dict : <EOL> grouped = self . _grouper ( rtn , <NUM_LIT:2> ) <EOL> rtn = dict ( ( a , b ) for a , b in grouped ) <EOL> return rtn <EOL> def pause ( self , queue_name , kw_in = None , kw_out = None , kw_all = None , <EOL> kw_none = None , kw_state = None , kw_bcast = None ) : <EOL> """<STR_LIT>""" <EOL> command = [ "<STR_LIT>" , queue_name ] <EOL> if kw_in : <EOL> command += [ "<STR_LIT>" ] <EOL> if kw_out : <EOL> command += [ "<STR_LIT>" ] <EOL> if kw_all : <EOL> command += [ "<STR_LIT:all>" ] <EOL> if kw_none : <EOL> command += [ "<STR_LIT:none>" ] <EOL> if kw_state : <EOL> command += [ "<STR_LIT:state>" ] <EOL> if kw_bcast : <EOL> command += [ "<STR_LIT>" ] <EOL> return self . execute_command ( * command ) <EOL> def qscan ( self , cursor = <NUM_LIT:0> , count = None , busyloop = None , minlen = None , <EOL> maxlen = None , importrate = None ) : <EOL> """<STR_LIT>""" <EOL> command = [ "<STR_LIT>" , cursor ] <EOL> if count : <EOL> command += [ "<STR_LIT>" , count ] <EOL> if busyloop : <EOL> command += [ "<STR_LIT>" ] <EOL> if minlen : <EOL> command += [ "<STR_LIT>" , minlen ] <EOL> if maxlen : <EOL> command += [ "<STR_LIT>" , maxlen ] <EOL> if importrate : <EOL> command += [ "<STR_LIT>" , importrate ] <EOL> return self . execute_command ( * command ) <EOL> def jscan ( self , cursor = <NUM_LIT:0> , count = None , busyloop = None , queue = None , <EOL> state = None , reply = None ) : <EOL> """<STR_LIT>""" <EOL> command = [ "<STR_LIT>" , cursor ] <EOL> if count : <EOL> command += [ "<STR_LIT>" , count ] <EOL> if busyloop : <EOL> command += [ "<STR_LIT>" ] <EOL> if queue : <EOL> command += [ "<STR_LIT>" , queue ] <EOL> if type ( state ) is list : <EOL> for s in state : <EOL> command += [ "<STR_LIT>" , s ] <EOL> if reply : <EOL> command += [ "<STR_LIT>" , reply ] <EOL> return self . execute_command ( * command ) <EOL> def hello ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . execute_command ( "<STR_LIT>" ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> logging . basicConfig ( level = logging . DEBUG ) <EOL> c = Client ( [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> c . connect ( ) <EOL> import json <EOL> job = json . dumps ( [ "<STR_LIT:hello>" , "<STR_LIT>" ] ) <EOL> logger . info ( c . add_job ( "<STR_LIT:test>" , job ) ) <EOL> jobs = c . get_job ( [ '<STR_LIT:test>' ] , timeout = <NUM_LIT:5> ) <EOL> for queue_name , job_id , payload in jobs : <EOL> logger . info ( job_id ) <EOL> c . ack_job ( job_id ) </s>
