added some fixes and the web interface for

delayed tasks

Author: Matt George

pyres/__init__.py

@@ -177,24 +177,27 @@ def close(self):
         """
         self.redis.disconnect()
 
-    def enqueue_at(self, timestamp, klass, *args):
+    def enqueue_at(self, datetime, klass, *args):
         class_name = '%s.%s' % (klass.__module__, klass.__name__)
-        self.delayed_push(timestamp, {'class':class_name,'queue': klass.queue, 'args':args})
+        self.delayed_push(datetime, {'class':class_name,'queue': klass.queue, 'args':args})
 
-    def delayed_push(self, timestamp, item):
-        key = int(time.mktime(timestamp.timetuple()))
+    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)]
+        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:%i" % key)
+        #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()))

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.get_type('resque:'+key)
     item = 0

tests/test_schedule.py

@@ -34,9 +34,9 @@ def test_delayed_timestamp_size(self):
         key = int(time.mktime(d.timetuple()))
         key2 = int(time.mktime(d2.timetuple()))
         self.resq.enqueue_at(d, Basic,"test1")
-        assert self.resq.delayed_timestamp_size(d) == 1
+        assert self.resq.delayed_timestamp_size(key) == 1
         self.resq.enqueue_at(d, Basic,"test1")
-        assert self.resq.delayed_timestamp_size(d) == 2
+        assert self.resq.delayed_timestamp_size(key) == 2
 
     def test_next_delayed_timestamp(self):
         d = datetime.datetime.now() + datetime.timedelta(days=-1)