adding the basics, still quite rough

Author: Matt George

README.markdown

@@ -0,0 +1 @@
+multiprocessing==2.6.2.1

requirements.txt

@@ -0,0 +1,11 @@
+from distutils.core import setup
+
+setup(name='PyRes',
+      version='0.1',
+      description='Python Resque Job',
+      author='Matt George',
+      author_email='[email protected]',
+      url='',
+      packages=['pyres'],
+      package_dir = {'': 'src'}
+)

setup.py

@@ -0,0 +1,97 @@
+from redis import Redis
+import simplejson
+
+import types
+def str_to_class(s):
+    lst = s.split(".")
+    klass = lst[-1]
+    mod_list = lst[:-1]
+    module = ".".join(mod_list)
+    try:
+        mod = __import__(module)
+        if hasattr(mod, klass):
+            return getattr(mod, klass)
+        else:
+            return None
+    except ImportError:
+        return None
+
+class ResQ(object):
+    _res = None
+    def __init__(self, server="localhost:6379"):
+        host, port = server.split(':')
+        self._redis = Redis(host=host, port=int(port))
+        self._watched_queues = {}
+    
+    def push(self, queue, item):
+        self.watch_queue(queue)
+        self._redis.push("queue:%s" % queue, ResQ.encode(item))
+    
+    def pop(self, queue):
+        ret = self._redis.pop("queue:%s" % queue)
+        if ret:
+            return ResQ.decode(ret)
+        return ret
+    
+    def size(self, queue):
+        return int(self._redis.llen("queue:%s" % queue))
+    
+    def watch_queue(self, queue):
+        if self._watched_queues.has_key(queue):
+            return
+        else:
+            if self._redis.sadd('queues',str(queue)):
+                self._watched_queues[queue] = queue
+    def peek(self, queue, start=0, count=1):
+        return self.list_range('queue:%s' % queue, start, count)
+    
+    def list_range(self, key, start, count):
+        items = self._redis.lrange(key,start,start+count-1)
+        ret_list = []
+        for i in items:
+            ret_list.append(ResQ.decode(i))
+        return ret_list
+    @classmethod
+    def encode(cls, item):
+        return simplejson.dumps(item)
+    
+    @classmethod
+    def decode(cls, item):
+        ret = simplejson.loads(item)
+        return ret
+    
+    @classmethod
+    def enqueue(cls, klass, *args):
+        queue = getattr(klass,'queue', None)
+        #print cls._res
+        resq = cls()
+        if queue:
+            class_name = '%s.%s' % (klass.__module__, klass.__name__)
+            #print class_name
+            resq.push(queue, {'klass':class_name,'args':args})
+            #Job.create(queue, klass,*args)
+    
+    @classmethod
+    def queues(cls, server="localhost:6379"):
+        resq = cls(server)
+        return resq._redis.smembers("queues")
+    
+class Stat(object):
+    def __init__(self, name, resq):
+        self.name = name
+        self.key = "stat:%s" % self.name
+        self.resq = resq
+    
+    def get(self):
+        val = self.resq._redis.get(self.key)
+        return int(val)
+    
+    def incr(self, ammount=1):
+        self.resq._redis.incr(self.key, ammount)
+    
+    def decr(self, ammount=1):
+        self.resq._redis.decr(self.key, ammount)
+    
+    def clear(self):
+        self.resq._redis.delete(self.key)
+    

src/pyres/__init__.py

@@ -0,0 +1,2 @@
+class NoQueueError(Exception):
+    pass

src/pyres/exceptions.py

@@ -0,0 +1,27 @@
+import datetime
+from pyres import ResQ
+class Failure(object):
+    def __init__(self, exp, queue, payload):
+        self._exception = exp
+        #self._worker = worker
+        self._queue = queue
+        self._payload = payload
+    
+    def save(self, resq):
+        data = {
+            'failed_at' : str(datetime.datetime.now()),
+            'payload'   : self._payload,
+            'error'     : self._exception,
+            'queue'     : self._queue
+        }
+        data = ResQ.encode(data)
+        resq._redis.push('failed', data)
+    
+    @classmethod
+    def count(cls, resq):
+        return int(resq._redis.llen('failed'))
+    
+    @classmethod
+    def create(cls, options={}):
+        pass
+    

src/pyres/failure.py

@@ -0,0 +1,28 @@
+from pyres import ResQ, str_to_class
+from pyres.failure import Failure
+class Job(object):
+    def __init__(self, queue, payload, resq):
+        self._queue = queue
+        self._payload = payload
+        self.resq = resq
+    
+    def perform(self):
+        payload_class_str = self._payload["klass"]
+        payload_class = str_to_class(payload_class_str)
+        args = self._payload.get("args", None)
+        if args:
+            return payload_class.perform(*args)
+        else:
+            return payload_class.perform()
+    
+    def fail(self, exception):
+        #Failure.create(exception)
+        failure = Failure(exception, self._queue, self._payload)
+        failure.save(self.resq)
+    
+    @classmethod
+    def reserve(cls, queue, res):
+        payload = res.pop(queue)
+        print "payload: %s" % payload
+        if payload:
+            return cls(queue, payload, res)

src/pyres/job.py

@@ -0,0 +1,61 @@
+from pyres.exceptions import NoQueueError
+class Manager(object):
+    def __init__(self, queues, host):
+        self.queues = queues
+        self._host = host
+        self.resq = ResQ(host)
+        self.validate_queues()
+    
+    def __str__(self):
+        import os; 
+        hostname = os.uname()[1]
+        pid = os.getpid()
+        return 'Manager:%s:%s:%s' % (hostname, pid, ','.join(self.queues))
+    
+    def validate_queues(self):
+        if not self.queues:
+            raise NoQueueError("Please give each worker at least one queue.")
+    
+    def work(self):
+        self.startup()
+        while True:
+            pass
+    
+    def startup(self):
+        self.register_manager()
+        self.register_signals()
+    
+    def register_manager(self):
+        self.resq._redis.sadd('managers',str(self))
+    
+    def unregister_manager(self):
+        self.resq._redis.srem('managers',str(self))
+    
+    def register_signals(self):
+        signal.signal(signal.SIGTERM, self.shutdown_all)
+        signal.signal(signal.SIGINT, self.shutdown_all)
+        signal.signal(signal.SIGQUIT, self.schedule_shutdown)
+        signal.signal(signal.SIGUSR1, self.kill_children)
+    
+    def shutdown_all(self, signum, frame):
+        self.schedule_shutdown(signum, frame)
+        self.kill_children(signum, frame)
+    
+    def schedule_shutdown(self, signum, frame):
+        self._shutdown = True
+    
+    def kill_children(self):
+        for child in self._children:
+            child.terminate()
+    
+    def start_child(self, queue):
+        from pyres.worker import JuniorWorker
+        p = Process(target=JuniorWorker.run, args=([queue], self._host))
+        self.children.append(p)
+        p.start()
+        return True
+    
+    @classmethod
+    def run(cls, queues=[], host="localhost:6379"):
+        manager = cls(queues, host)
+        manager.work()