Merge branch 'scheduled_tasks'

Conflicts:
	pyres/__init__.py

Author: Matt George

pyres/__init__.py

@@ -3,6 +3,8 @@
 from redis import Redis
 import pyres.json_parser as json
 
+import time, datetime
+
 import logging
 
 def my_import(name):
@@ -19,12 +21,12 @@ def safe_str_to_class(s):
     klass = lst[-1]
     mod_list = lst[:-1]
     module = ".".join(mod_list)
-    mod = my_import(module)
-    if hasattr(mod, klass):
-        return getattr(mod, klass)
-    else:
+        mod = my_import(module)
+        if hasattr(mod, klass):
+            return getattr(mod, klass)
+        else:
         raise ImportError('')
-    
+
 def str_to_class(s):
     """Alternate helper function to map string class names to module classes."""
     lst = s.split(".")
@@ -199,6 +201,51 @@ def close(self):
         """
         self.redis.disconnect()
 
+    def enqueue_at(self, datetime, klass, *args):
+        class_name = '%s.%s' % (klass.__module__, klass.__name__)
+        logging.info("enqueued '%s' job for execution at %s" % (class_name, datetime))
+        if args:
+            logging.debug("job arguments are: %s" % str(args))
+        self.delayed_push(datetime, {'class':class_name,'queue': klass.queue, 'args':args})
+    
+    def delayed_push(self, datetime, item):
+        key = int(time.mktime(datetime.timetuple()))
+        self.redis.rpush('resque:delayed:%s' % key, ResQ.encode(item))
+        self.redis.zadd('resque:delayed_queue_schedule', key, key)
+    
+    def delayed_queue_peek(self, start, count):
+        return [int(item) for item in self.redis.zrange('resque:delayed_queue_schedule', start, start+count) or []]
+    
+    def delayed_timestamp_peek(self, timestamp, start, count):
+        return self.list_range('resque:delayed:%s' % timestamp, start, count)
+        
+    def delayed_queue_schedule_size(self):
+        return self.redis.zcard('resque:delayed_queue_schedule')
+    
+    def delayed_timestamp_size(self, timestamp):
+        #key = int(time.mktime(timestamp.timetuple()))
+        return self.redis.llen("resque:delayed:%s" % timestamp)
+    
+    def next_delayed_timestamp(self):
+        key = int(time.mktime(datetime.datetime.now().timetuple()))
+        array = self.redis.zrangebyscore('resque:delayed_queue_schedule', '-inf', key)
+        timestamp = None
+        if array:
+            timestamp = array[0]
+        return timestamp
+    
+    def next_item_for_timestamp(self, timestamp):
+        #key = int(time.mktime(timestamp.timetuple()))
+        key = "resque:delayed:%s" % timestamp
+        ret = self.redis.lpop(key)
+        item = None
+        if ret:
+            item = ResQ.decode(ret)
+        if self.redis.llen(key) == 0:
+            self.redis.delete(key)
+            self.redis.zrem('resque:delayed_queue_schedule', timestamp)
+        return item
+    
     @classmethod
     def encode(cls, item):
         return json.dumps(item)

pyres/scheduler.py

@@ -0,0 +1,78 @@
+import signal
+import time
+import logging
+
+from pyres import ResQ
+
+class Scheduler(object):
+    
+    def __init__(self, server="localhost:6379", password=None):
+        """
+        >>> from pyres.scheduler import Scheduler
+        >>> scheduler = Scheduler('localhost:6379')
+        """
+        self._shutdown = False
+        if isinstance(server,basestring):
+            self.resq = ResQ(server=server, password=password)
+        elif isinstance(server, ResQ):
+            self.resq = server
+        else:
+            raise Exception("Bad server argument")
+    
+    def register_signal_handlers(self):
+        logging.info('registering signals')
+        signal.signal(signal.SIGTERM, self.schedule_shutdown)
+        signal.signal(signal.SIGINT, self.schedule_shutdown)
+        signal.signal(signal.SIGQUIT, self.schedule_shutdown)
+    
+    def schedule_shutdown(self, signal, frame):
+        logging.info('shutting down started')
+        self._shutdown = True
+    
+    def __call__(self):
+        logging.info('starting up')
+        self.register_signal_handlers()
+        #self.load_schedule()
+        logging.info('looking for delayed items')
+        while True:
+            if self._shutdown is True:
+                break
+            self.handle_delayed_items()
+            logging.debug('sleeping')
+            time.sleep(5)
+        logging.info('shutting down complete')
+    
+    def next_timestamp(self):
+        while True:
+            timestamp = self.resq.next_delayed_timestamp()
+            if timestamp:
+                yield timestamp
+            else:
+                break
+            
+    
+    def next_item(self, timestamp):
+        while True:
+            item = self.resq.next_item_for_timestamp(timestamp)
+            if item:
+                yield item
+            else:
+                break
+    
+    def handle_delayed_items(self):
+        for timestamp in self.next_timestamp():
+            logging.info('handling timestamp: %s' % timestamp)
+            for item in self.next_item(timestamp):
+                logging.debug('queueing item %s' % item)
+                klass = item['class']
+                queue = item['queue']
+                args = item['args']
+                self.resq.enqueue_from_string(klass, queue, *args)
+            
+        
+    @classmethod
+    def run(cls, server, password=None):
+        sched = cls(server=server, password=password)
+        sched()
+    
+    

resweb/server.py

@@ -11,7 +11,9 @@
     Failed, 
     Stats, 
     Stat, 
-    Worker
+    Worker,
+    Delayed,
+    DelayedTimestamp
 )
 
 HOST = ResQ("localhost:6379")
@@ -61,6 +63,18 @@ def stats(request, key):
 def stat(request, stat_id):
     return str(Stat(HOST, stat_id).render())
 
+@get('/delayed/')
+def delayed(request):
+    start = request.GET.get('start',0)
+    start = int(start)
+    return str(Delayed(HOST, start).render())
+
+@get('/delayed/(?P<timestamp>\w.+)')
+def delayed_timestamp(request, timestamp):
+    start = request.GET.get('start',0)
+    start = int(start)
+    return str(DelayedTimestamp(HOST, timestamp, start).render())
+
 @get('/media/(?P<filename>.+)')
 def my_media(request, filename):
     print filename

resweb/templates/delayed.mustache

@@ -0,0 +1,31 @@
+{{>header}}
+<h1>Delayed Jobs</h1>
+
+<p class='intro'>
+  This list below contains the timestamps for scheduled delayed jobs.
+</p>
+
+<p class='sub'>
+  Showing {{start}} to {{end}} of <b>{{size}}</b> timestamps
+</p>
+
+
+<table>
+  <tr>
+    <th>Timestamp</th>
+    <th>Job count</th>
+  </tr>
+  {{#jobs}}
+    <tr>
+      <td><a href="/delayed/{{timestamp}}/">{{formated_time}}</a></td>
+      <td>{{size}}</td>
+    </tr>
+  {{/jobs}}
+</table>
+
+{{#pagination}}
+  <li>
+      {{#current}}<a href='{{link}}'>{{/current}}{{link_name}}{{#current}}</a>{{/current}}
+  </li>
+{{/pagination}}
+{{>footer}}

resweb/templates/delayed_timestamp.mustache

@@ -0,0 +1,30 @@
+{{>header}}
+
+<h1>Delayed jobs scheduled for {{formated_timestamp}}</h1>
+
+<p class='sub'>Showing {{start}} to {{end}} of <b>{{size}}</b> jobs</p>
+
+<table class='jobs'>
+  <tr>
+    <th>Class</th>
+    <th>Args</th>
+  </tr>
+  {{#jobs}}
+    <tr>
+      <td class='class'>{{class}}</td>
+      <td class='args'>{{args}}</td>
+    </tr>
+  {{/jobs}}
+  {{#no_jobs}}
+    <tr>
+      <td class='no-data' colspan='2'>There are no pending jobs scheduled for this time.</td>
+    </tr>
+  {{/no_jobs}}
+</table>
+
+{{#pagination}}
+  <li>
+      {{#current}}<a href='{{link}}'>{{/current}}{{link_name}}{{#current}}</a>{{/current}}
+  </li>
+{{/pagination}}
+{{>footer}}

resweb/templates/header.mustache

@@ -14,6 +14,7 @@
         <li><a href='/failed/'>Failed</a></li>
         <li><a href='/queues/'>Queues</a></li>
         <li><a href='/workers/'>Workers</a></li>
+        <li><a href='/delayed/'>Delayed</a></li>
         <li><a href='/stats/'>Stats</a></li>
     </ul>
   </div>

resweb/views.py

@@ -1,24 +1,30 @@
 import pystache
 
-from pyres import ResQ, __version__
+from pyres import __version__
 from pyres.worker import Worker as Wrkr
 from pyres import failure
 import os
+import datetime
+
 TEMPLATE_PATH = os.path.join(os.path.dirname(__file__), 'templates')
 class ResWeb(pystache.View):
     template_path = TEMPLATE_PATH
     def __init__(self, host):
         super(ResWeb, self).__init__()
         self.resq = host
+    
     def media_folder(self):
         return '/media/'
+    
     def close(self):
         self.resq.close()
 
     def address(self):
         return '%s:%s' % (self.resq.redis.host,self.resq.redis.port)
+    
     def version(self):
         return str(__version__)
+    
     def pages(self, start, size, link_function, width=20):
         pages = []
 
@@ -406,6 +412,80 @@ def runat(self):
         item['nodata'] = not item['data']
         """
         pass
+
+class Delayed(ResWeb):
+    def __init__(self, host, start=0):
+        self._start = start
+        super(Delayed, self).__init__(host)
+
+    def start(self):
+        return str(self._start)
+
+    def end(self):
+        return str(self._start + 20)
+
+    def size(self):
+        item = self.resq.delayed_queue_schedule_size() or 0
+        return str(item)
+
+    def jobs(self):
+        jobs = []
+        for timestamp in self.resq.delayed_queue_peek(self.start(), self.end()):
+            t = datetime.datetime.fromtimestamp(float(timestamp))
+            item = dict(timestamp=str(timestamp))
+            item['size'] = str(self.resq.delayed_timestamp_size(timestamp))
+            
+            item['formated_time'] = str(t)
+            
+            jobs.append(item)
+        return jobs
+
+    def pagination(self):
+        return self.pages(self._start, int(self.size()), self.link_func)
+
+    def link_func(self, start):
+        return '/delayed/?start=%s' % start
+
+class DelayedTimestamp(ResWeb):
+    def __init__(self, host, timestamp, start=0):
+        self._start = start
+        self._timestamp = timestamp
+        super(DelayedTimestamp, self).__init__(host)
+    
+    def formated_timestamp(self):
+        return str(datetime.datetime.fromtimestamp(float(self._timestamp)))
+    
+    def start(self):
+        return str(self._start)
+
+    def end(self):
+        return str(self._start + 20)
+
+    def size(self):
+        item = self.resq.delayed_timestamp_size(self._timestamp) or 0
+        return str(item)
+
+    def jobs(self):
+        jobs = []
+        for job in self.resq.delayed_timestamp_peek(self._timestamp, int(self.start()), int(self.end())):
+            item = {
+                'class': str(job['class']),
+                'args': str(job['args'])
+            }
+            jobs.append(item)
+        return jobs
+    
+    def no_jobs(self):
+        if int(self.size()) > 0:
+            return False
+        return True
+    
+    def pagination(self):
+        return self.pages(self._start, int(self.size()), self.link_func)
+
+    def link_func(self, start):
+        return '/delayed/?start=%s' % start
+
 def redis_size(key, resq):
     key_type = resq.redis.type('resque:'+key)
     item = 0

scripts/pyres_scheduler

@@ -0,0 +1,20 @@
+#!/usr/bin/env python
+from pyres.scheduler import Scheduler
+import logging
+
+def main():
+    from optparse import OptionParser
+    usage = "usage: %prog [options] arg1"
+    parser = OptionParser(usage=usage)
+    #parser.add_option("-q", dest="queue_list")
+    parser.add_option("--host", dest="host", default="localhost")
+    parser.add_option("--port",dest="port",type="int", default=6379)
+    parser.add_option('-l', '--log-level', dest='log_level', default='info', help='log level.  Valid values are "debug", "info", "warning", "error", "critical", in decreasing order of verbosity. Defaults to "info" if parameter not specified.')
+    (options,args) = parser.parse_args()
+    log_level = getattr(logging, options.log_level.upper(),'INFO')
+    logging.basicConfig(level=log_level, format="%(module)s: %(asctime)s: %(levelname)s: %(message)s")
+    server = '%s:%s' % (options.host, options.port)
+    Scheduler.run(server)
+
+if __name__ == "__main__":
+    main()