Merge branch 'failure_backend_support'

Author: Matt George

pyres/__init__.py

@@ -75,7 +75,7 @@ def peek(self, queue, start=0, count=1):
         return self.list_range('resque:queue:%s' % queue, start, count)
 
     def list_range(self, key, start, count):
-        items = self.redis.lrange(key,start,start+count-1)
+        items = self.redis.lrange(key, start,start+count-1)
         ret_list = []
         for i in items:
             ret_list.append(ResQ.decode(i))

pyres/failure/__init__.py

@@ -0,0 +1,22 @@
+import datetime
+from pyres import ResQ 
+import sys, traceback
+from pyres.failure.redis import RedisBackend
+_backend = RedisBackend
+def create(*args, **kwargs):
+    return _backend(*args, **kwargs)
+
+def count(resq):
+    return _backend.count(resq)
+
+def all(resq, start, count):
+    return _backend.all(resq, start, count)
+
+def clear(resq):
+    return _backend.clear(resq)
+
+def requeue(resq, failure_object):
+    queue = failure_object._queue
+    payload = failure_object._payload
+    return resq.push(queue,payload)
+    

pyres/failure.py pyres/failure/base.pypyres/failure.py renamed to pyres/failure/base.py

@@ -1,7 +1,7 @@
-import datetime
-from pyres import ResQ 
-import sys, traceback
-class Failure(object):
+import sys
+import traceback
+
+class BaseBackend(object):
     def __init__(self, exp, queue, payload, worker=None):
         excc, _, tb = sys.exc_info()
 
@@ -11,6 +11,7 @@ def __init__(self, exp, queue, payload, worker=None):
         self._queue = queue
         self._payload = payload
 
+    
     def _parse_traceback(self, trace):
         """Return the given traceback string formatted for a notification."""
         p_traceback = [ "%s:%d:in `%s'" % (filename, lineno, funcname) 
@@ -23,21 +24,4 @@ def _parse_traceback(self, trace):
     def _parse_message(self, exc):
         """Return a message for a notification from the given exception."""
         return '%s: %s' % (exc.__class__.__name__, str(exc))
-    
-    def save(self, resq):
-        data = {
-            'failed_at' : str(datetime.datetime.now()),
-            'payload'   : self._payload,
-            'error'     : self._parse_message(self._exception),
-            'backtrace' : self._parse_traceback(self._traceback),
-            'queue'     : self._queue,
-            'worker'    : self._worker
-        }
-        data = ResQ.encode(data)
-        resq.redis.push('resque:failed', data)
-    
-    @classmethod
-    def count(cls, resq):
-        return int(resq.redis.llen('resque:failed'))
-    
-    
+    

pyres/failure/redis.py

@@ -0,0 +1,31 @@
+import datetime
+from base import BaseBackend
+from pyres import ResQ
+class RedisBackend(BaseBackend):
+    def save(self, resq=None):
+        if not resq:
+            resq = ResQ()
+        data = {
+            'failed_at' : str(datetime.datetime.now()),
+            'payload'   : self._payload,
+            'error'     : self._parse_message(self._exception),
+            'backtrace' : self._parse_traceback(self._traceback),
+            'queue'     : self._queue
+        }
+        if self._worker:
+            data['worker'] = self._worker
+        data = ResQ.encode(data)
+        resq.redis.push('resque:failed', data)
+    
+    @classmethod
+    def count(cls, resq):
+        return int(resq.redis.llen('resque:failed'))
+    
+    @classmethod
+    def all(cls, resq, start=0, count=1):
+        return resq.list_range('resque:failed', start, count)
+        
+    @classmethod
+    def clear(cls, resq):
+        return resq.redis.delete('resque:failed')
+    

pyres/job.py

@@ -1,5 +1,5 @@
 from pyres import ResQ, str_to_class, safe_str_to_class
-from pyres.failure import Failure
+from pyres import failure
 class Job(object):
     def __init__(self, queue, payload, resq, worker=None):
         self._queue = queue
@@ -18,8 +18,9 @@ def perform(self):
 
     def fail(self, exception):
         #Failure.create(exception)
-        failure = Failure(exception, self._queue, self._payload, self._worker)
-        failure.save(self.resq)
+        fail = failure.create(exception, self._queue, self._payload, self._worker)
+        fail.save(self.resq)
+        return fail
 
     @classmethod
     def reserve(cls, queue, res, worker=None):

resweb/server.py

@@ -1,7 +1,7 @@
 import os
 from itty import *
 from pyres import ResQ
-from pyres.failure import Failure
+from pyres import failure
 from views import (
     Overview, 
     Queues, 
@@ -36,7 +36,9 @@ def queue(request, queue_id):
 
 @get('/failed/')
 def failed(request):
-    return str(Failed(HOST).render())
+    start = request.GET.get('start',0)
+    start = int(start)
+    return str(Failed(HOST, start).render())
 
 @get('/workers/(?P<worker_id>\w.+)/')
 def worker(request, worker_id):
@@ -69,4 +71,5 @@ def my_media(request, filename):
     #return Response(output, content_type=content_type(filename))
     #return static_file(request, filename=filename, root=my_root)
 
-#run_itty()
+if __name__ == "__main__":
+    run_itty()

resweb/templates/failed.mustache

@@ -1,4 +1,38 @@
 {{>header}}
+<script type="text/javascript">
+$(document).ready(function(){
+    $('.backtrace').click(function() {
+        $(this).next().toggle();
+        return false;
+    });
+});
+</script>
 <h1>Failed Jobs</h1>
-Pending Failure module completion!
-{{>footer}}
+<p class='sub'>Showing {{start}} to {{end}} of <b>{{size}}</b> jobs</p>
+
+<ul class='failed'>
+    {{#failed_jobs}}
+    <li>
+      <dl>
+        <dt>Worker</dt>
+        <dd>
+            <a href="{{worker_url}}">{{worker}}</a> on <b class='queue-tag'>{{queue}}</b > at <b>
+            <span class="time">{{failed_at}}</span></b>
+        </dd>
+        <dt>Class</dt>
+        <dd><code>{{payload_class}}</code></dd>
+        <dt>Arguments</dt>
+        <dd><pre>{{payload_args}}</pre></dd>
+        <dt>Error</dt>
+        <dd>
+          <a href="#" class="backtrace">{{error}}</a>
+          <pre style='display:none'>{{traceback}}</pre>          
+        </dd>
+      </dl>
+      <div class='r'>
+      </div>
+    </li>
+  {{/failed_jobs}}
+</ul>
+
+{{>footer}}

resweb/templates/header.mustache

@@ -4,7 +4,7 @@
   <title>Resque.</title>
   <link href="{{media_folder}}reset.css" media="screen" rel="stylesheet" type="text/css">
   <link href="{{media_folder}}style.css" media="screen" rel="stylesheet" type="text/css">
-  
+  <script src="{{media_folder}}jquery-1.3.2.min.js" type="text/javascript"</script>
 </head>
 <body>
   <div class="header">

resweb/views.py

@@ -2,6 +2,7 @@
 
 from pyres import ResQ, __version__
 from pyres.worker import Worker as Wrkr
+from pyres import failure
 import os
 TEMPLATE_PATH = os.path.join(os.path.dirname(__file__), 'templates')
 class ResWeb(pystache.View):
@@ -60,8 +61,8 @@ def empty(self):
         return not self._queue
 
     def fail_count(self):
-        from pyres.failure import Failure
-        return str(Failure.count(self.resq))
+        #from pyres.failure import Failure
+        return str(failure.count(self.resq))
 
     def workers(self):
         workers = []
@@ -161,8 +162,25 @@ def __init__(self, host, start=0):
         self._start = start
         super(Failed, self).__init__(host)
 
+    def start(self):
+        return str(self._start)
+    
+    def end(self):
+        return str(self._start + 20)
+    
+    def size(self):
+        return str(failure.count(self.resq) or 0)
+     
     def failed_jobs(self):
-        return ''
+        jobs = []
+        for job in failure.all(self.resq, self._start, self._start + 20):
+            item = job
+            item['worker_url'] = '/workers/%s/' % job['worker']
+            item['payload_args'] = ','.join(job['payload']['args'])
+            item['payload_class'] = job['payload']['class']
+            item['traceback'] = '\n'.join(job['backtrace'])
+            jobs.append(item)
+        return jobs
 
 class Stats(ResWeb):
     def __init__(self, host, key_id):
@@ -354,4 +372,4 @@ def redis_size(key, resq):
         item = resq.redis.scard('resque:'+key)
     elif key_type == 'string':
         item = 1
-    return str(item)
+    return str(item)

tests/test_jobs.py

@@ -6,6 +6,7 @@ def test_reserve(self):
         job = Job.reserve('basic', self.resq)
         assert job._queue == 'basic'
         assert job._payload
+        assert job._payload == {'class':'tests.Basic','args':['test1']}
 
     def test_perform(self):
         self.resq.enqueue(Basic,"test1")