diff --git a/scrapyd/interfaces.py b/scrapyd/interfaces.py index bac4aa169..b64946968 100644 --- a/scrapyd/interfaces.py +++ b/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. diff --git a/scrapyd/spiderqueue.py b/scrapyd/spiderqueue.py index 672a549ac..faa706385 100644 --- a/scrapyd/spiderqueue.py +++ b/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() diff --git a/scrapyd/sqlite.py b/scrapyd/sqlite.py index d205d4508..0cbc7217b 100644 --- a/scrapyd/sqlite.py +++ b/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() diff --git a/scrapyd/tests/test_sqlite.py b/scrapyd/tests/test_sqlite.py index d0a4f09ec..51d62048a 100644 --- a/scrapyd/tests/test_sqlite.py +++ b/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) diff --git a/scrapyd/webservice.py b/scrapyd/webservice.py index 79bc53f64..088253749 100644 --- a/scrapyd/webservice.py +++ b/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): diff --git a/scrapyd/website.py b/scrapyd/website.py index db23de51f..7f8c7791b 100644 --- a/scrapyd/website.py +++ b/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))