scrapyd: added new cancel.json api to cancel pending/running jobs

scrapyd/interfaces.py

@@ -73,6 +73,11 @@ class ISpiderQueue(Interface):
 
         This method can return a deferred. """
 
+    def remove(func):
+        """Remove all elements from the queue for which func(element) is true,
+        and return the number of removed elements.
+        """
+
     def clear():
         """Clear the queue.
 

scrapyd/spiderqueue.py

@@ -26,5 +26,8 @@ class SqliteSpiderQueue(object):
     def list(self):
         return [x[0] for x in self.q]
 
+    def remove(self, func):
+        return self.q.remove(func)
+
     def clear(self):
         self.q.clear()

scrapyd/sqlite.py

@@ -119,6 +119,20 @@ class SqlitePriorityQueue(object):
         self.conn.commit()
         return self.decode(msg)
 
+    def remove(self, func):
+        q = "select id, message from %s" % self.table
+        n = 0
+        for id, msg in self.conn.execute(q):
+            if func(self.decode(msg)):
+                q = "delete from %s where id=?" % self.table
+                c = self.conn.execute(q, (id,))
+                if not c.rowcount: # record vanished, so let's try again
+                    self.conn.rollback()
+                    return self.remove(func)
+                n += 1
+        self.conn.commit()
+        return n
+
     def clear(self):
         self.conn.execute("delete from %s" % self.table)
         self.conn.commit()

scrapyd/tests/test_sqlite.py

@@ -125,6 +125,20 @@ class SqlitePriorityQueueTest(unittest.TestCase):
         self.failUnlessEqual(len(self.q), 0)
         self.failUnlessEqual(list(self.q), [])
 
+    def test_remove(self):
+        self.failUnlessEqual(len(self.q), 0)
+        self.failUnlessEqual(list(self.q), [])
+        msg1 = "good message 1"
+        msg2 = "bad message 2"
+        msg3 = "good message 3"
+        msg4 = "bad message 4"
+        self.q.put(msg1)
+        self.q.put(msg2)
+        self.q.put(msg3)
+        self.q.put(msg4)
+        self.q.remove(lambda x: x.startswith("bad"))
+        self.failUnlessEqual(list(self.q), [(msg1, 0.0), (msg3, 0.0)])
+
     def test_types(self):
         for x in self.supported_values:
             self.q.put(x)

scrapyd/webservice.py

@@ -34,6 +34,25 @@ class Schedule(WsResource):
         self.root.scheduler.schedule(project, spider, **args)
         return {"status": "ok", "jobid": jobid}
 
+class Cancel(WsResource):
+
+    def render_POST(self, txrequest):
+        args = dict((k, v[0]) for k, v in txrequest.args.items())
+        project = args['project']
+        jobid = args['job']
+        signal = args.get('signal', 'TERM')
+        prevstate = None
+        queue = self.root.poller.queues[project]
+        c = queue.remove(lambda x: x["_job"] == jobid)
+        if c:
+            prevstate = "pending"
+        spiders = self.root.launcher.processes.values()
+        for s in spiders:
+            if s.job == jobid:
+                s.transport.signalProcess(signal)
+                prevstate = "running"
+        return {"status": "ok", "prevstate": prevstate}
+
 class AddVersion(WsResource):
 
     def render_POST(self, txrequest):

scrapyd/website.py

@@ -16,6 +16,7 @@ class Root(resource.Resource):
         self.app = app
         self.putChild('', Home(self))
         self.putChild('schedule.json', webservice.Schedule(self))
+        self.putChild('cancel.json', webservice.Cancel(self))
         self.putChild('addversion.json', webservice.AddVersion(self))
         self.putChild('listprojects.json', webservice.ListProjects(self))
         self.putChild('listversions.json', webservice.ListVersions(self))