Added persistent execution queue (based on SQLite), and a new 'queue' command to control it. Closes #198

docs/topics/commands.rst

@@ -173,13 +173,17 @@ start
 | Requires project: | *yes*            |
 +-------------------+------------------+
 
-Start Scrapy in server mode.
+Start Scrapy in server mode, which can be controlled by the :command:`queue`
+command.
 
 Usage example::
 
     $ scrapy start
     [ ... scrapy starts and stays idle waiting for spiders to get scheduled ... ]
 
+You can now schedule spiders to run using the :command:`queue` command. If
+there were spiders already enqueued, it will start crawling them.
+
 .. command:: list
 
 list
@@ -352,3 +356,35 @@ Example usage::
 
     $ scrapy runspider myspider.py
     [ ... spider starts crawling ... ]
+
+.. command:: queue
+
+queue
+-----
+
++-------------------+----------------------------------------------+
+| Syntax:           | ``scrapy queue <list|clear|add spider1 ..>`` |
++-------------------+----------------------------------------------+
+| Requires project: | *yes*                                        |
++-------------------+----------------------------------------------+
+
+Manage the execution queue of a Scrapy project.
+
+This command is meant to be used to control a Scrapy server started with the
+:command:`start` command.
+
+Example usage::
+
+    $ scrapy queue add example.com
+
+If there is a Scrapy server running (see :command:`start` command), it will
+start crawling the ``example.com`` spider. Otherwise, it will only get
+enqueued,, and it will start crawling once the Scrapy server is started.
+
+You can also view the spiders enqueued but not yet started::
+
+    $ scrapy queue list
+
+And clear the queue::
+
+    $ scrapy queue clear

scrapy/commands/queue.py

@@ -0,0 +1,46 @@
+from scrapy.command import ScrapyCommand
+from scrapy.utils.misc import load_object
+from scrapy.conf import settings
+
+class Command(ScrapyCommand):
+
+    requires_project = True
+    default_settings = {'LOG_ENABLED': False}
+
+    def syntax(self):
+        return "[options] <list|clear|add spider1 ..>"
+
+    def short_desc(self):
+        return "Control execution queue"
+
+    def add_options(self, parser):
+        ScrapyCommand.add_options(self, parser)
+        parser.add_option("--priority", dest="priority", type="float", default=0.0, \
+            help="priority to use for adding spiders")
+        parser.add_option("-a", "--arg", dest="spargs", action="append", default=[], \
+            help="spider arguments to use for adding spiders")
+
+    def run(self, args, opts):
+        if len(args) < 1:
+            return False
+        cmd = args[0]
+
+        botname = settings['BOT_NAME']
+        queue = load_object(settings['SERVICE_QUEUE'])().queue
+
+        if cmd == 'add':
+            if len(args) < 2:
+                return False
+            msg = dict(x for x in [x.split('=', 1) for x in opts.spargs])
+            for x in args[1:]:
+                msg.update(name=x)
+                queue.put(msg)
+                print "Added (priority=%s): %s" % (opts.priority, msg)
+        elif cmd == 'list':
+            for x, y in queue:
+                print "(priority=%s) %s" % (y, x)
+        elif cmd == 'clear':
+            queue.clear()
+            print "Cleared %s queue" % botname
+        else:
+            return False

scrapy/conf/default_settings.py

@@ -209,7 +209,8 @@ SCHEDULER_MIDDLEWARES_BASE = {
 
 SCHEDULER_ORDER = 'DFO'
 
-SERVICE_QUEUE = 'scrapy.core.queue.KeepAliveExecutionQueue'
+SERVICE_QUEUE = 'scrapy.contrib.queue.SqliteExecutionQueue'
+SERVICE_QUEUE_FILE = 'scrapy.db'
 
 SPIDER_MANAGER_CLASS = 'scrapy.contrib.spidermanager.SpiderManager'
 

scrapy/contrib/queue/__init__.py

@@ -0,0 +1,20 @@
+from scrapy.core.queue import ExecutionQueue
+from scrapy.utils.sqlite import JsonSqlitePriorityQueue
+from scrapy.conf import settings
+
+class SqliteExecutionQueue(ExecutionQueue):
+
+    queue_class = JsonSqlitePriorityQueue
+
+    def __init__(self, *a, **kw):
+        super(SqliteExecutionQueue, self).__init__(*a, **kw)
+        self.queue = JsonSqlitePriorityQueue(settings['SERVICE_QUEUE_FILE'])
+
+    def _append_next(self):
+        msg = self.queue.pop()
+        if msg:
+            name = msg.pop('name')
+            self.append_spider_name(name, **msg)
+
+    def is_finished(self):
+        return False

scrapy/tests/test_utils_sqlite.py

@@ -0,0 +1,141 @@
+import unittest
+from datetime import datetime
+from decimal import Decimal
+
+from scrapy.http import Request
+from scrapy.utils.sqlite import SqlitePriorityQueue, JsonSqlitePriorityQueue, \
+    PickleSqlitePriorityQueue, SqliteDict, JsonSqliteDict, PickleSqliteDict
+
+
+class SqliteDictTest(unittest.TestCase):
+
+    dict_class = SqliteDict
+    test_dict = {'hello': 'world', 'int': 1, 'float': 1.5}
+
+    def test_basic_types(self):
+        test = self.test_dict
+        d = self.dict_class()
+        d.update(test)
+        self.failUnlessEqual(d.items(), test.items())
+        d.clear()
+        self.failIf(d.items())
+
+
+class JsonSqliteDictTest(SqliteDictTest):
+
+    dict_class = JsonSqliteDict
+    test_dict = SqliteDictTest.test_dict.copy()
+    test_dict.update({'list': ['a', 'world'], 'dict': {'some': 'dict'}})
+
+
+class PickleSqliteDictTest(JsonSqliteDictTest):
+
+    dict_class = PickleSqliteDict
+    test_dict = JsonSqliteDictTest.test_dict.copy()
+    test_dict.update({'decimal': Decimal("10"), 'datetime': datetime.now()})
+
+    def test_request_persistance(self):
+        r1 = Request("http://www.example.com", body="some")
+        d = self.dict_class()
+        d['request'] = r1
+        r2 = d['request']
+        self.failUnless(isinstance(r2, Request))
+        self.failUnlessEqual(r1.url, r2.url)
+        self.failUnlessEqual(r1.body, r2.body)
+
+
+class SqlitePriorityQueueTest(unittest.TestCase):
+
+    queue_class = SqlitePriorityQueue
+
+    supported_values = ["bytes", u"\xa3", 123, 1.2, True]
+
+    def setUp(self):
+        self.q = self.queue_class()
+
+    def test_empty(self):
+        self.failUnless(self.q.pop() is None)
+
+    def test_one(self):
+        msg = "a message"
+        self.q.put(msg)
+        self.failIf("_id" in msg)
+        self.failUnlessEqual(self.q.pop(), msg)
+        self.failUnless(self.q.pop() is None)
+
+    def test_multiple(self):
+        msg1 = "first message"
+        msg2 = "second message"
+        self.q.put(msg1)
+        self.q.put(msg2)
+        out = []
+        out.append(self.q.pop())
+        out.append(self.q.pop())
+        self.failUnless(msg1 in out)
+        self.failUnless(msg2 in out)
+        self.failUnless(self.q.pop() is None)
+
+    def test_priority(self):
+        msg1 = "message 1"
+        msg2 = "message 2"
+        msg3 = "message 3"
+        msg4 = "message 4"
+        self.q.put(msg1, priority=1.0)
+        self.q.put(msg2, priority=5.0)
+        self.q.put(msg3, priority=3.0)
+        self.q.put(msg4, priority=2.0)
+        self.failUnlessEqual(self.q.pop(), msg2)
+        self.failUnlessEqual(self.q.pop(), msg3)
+        self.failUnlessEqual(self.q.pop(), msg4)
+        self.failUnlessEqual(self.q.pop(), msg1)
+
+    def test_iter_len_clear(self):
+        self.failUnlessEqual(len(self.q), 0)
+        self.failUnlessEqual(list(self.q), [])
+        msg1 = "message 1"
+        msg2 = "message 2"
+        msg3 = "message 3"
+        msg4 = "message 4"
+        self.q.put(msg1, priority=1.0)
+        self.q.put(msg2, priority=5.0)
+        self.q.put(msg3, priority=3.0)
+        self.q.put(msg4, priority=2.0)
+        self.failUnlessEqual(len(self.q), 4)
+        self.failUnlessEqual(list(self.q), \
+            [(msg2, 5.0), (msg3, 3.0), (msg4, 2.0), (msg1, 1.0)])
+        self.q.clear()
+        self.failUnlessEqual(len(self.q), 0)
+        self.failUnlessEqual(list(self.q), [])
+
+    def test_types(self):
+        for x in self.supported_values:
+            self.q.put(x)
+            self.failUnlessEqual(self.q.pop(), x)
+
+
+class JsonSqlitePriorityQueueTest(SqlitePriorityQueueTest):
+
+    queue_class = JsonSqlitePriorityQueue
+
+    supported_values = SqlitePriorityQueueTest.supported_values + [
+        ["a", "list", 1],
+        {"a": "dict"},
+    ]
+
+
+class PickleSqlitePriorityQueueTest(JsonSqlitePriorityQueueTest):
+
+    queue_class = PickleSqlitePriorityQueue
+
+    supported_values = JsonSqlitePriorityQueueTest.supported_values + [
+        Decimal("10"),
+        datetime.now(),
+    ]
+
+    def test_request_persistance(self):
+        r1 = Request("http://www.example.com", body="some")
+        self.q.put(r1)
+        r2 = self.q.pop()
+        self.failUnless(isinstance(r2, Request))
+        self.failUnlessEqual(r1.url, r2.url)
+        self.failUnlessEqual(r1.body, r2.body)

scrapy/utils/sqlite.py

@@ -0,0 +1,155 @@
+import sqlite3
+import cPickle
+from UserDict import DictMixin
+
+from scrapy.utils.py26 import json
+
+
+class SqliteDict(DictMixin):
+    """SQLite-backed dictionary"""
+
+    def __init__(self, database=':memory:', table="dict"):
+        self.database = database
+        self.table = table
+        self.conn = sqlite3.connect(database)
+        q = "create table if not exists %s (key text primary key, value blob)" \
+            % table
+        self.conn.execute(q)
+
+    def __getitem__(self, key):
+        key = self.encode(key)
+        q = "select value from %s where key=?" % self.table
+        value = self.conn.execute(q, (key,)).fetchone()[0]
+        return self.decode(value)
+
+    def __setitem__(self, key, value):
+        key, value = self.encode(key), self.encode(value)
+        q = "insert into %s (key, value) values (?,?)" % self.table
+        self.conn.execute(q, (key, value))
+        self.conn.commit()
+
+    def __delitem__(self, key):
+        key = self.encode(key)
+        q = "delete from %s where key=?" % self.table
+        self.conn.execute(q, (key,))
+        self.conn.commit()
+
+    def iterkeys(self):
+        q = "select key from %s" % self.table
+        return (self.decode(x[0]) for x in self.conn.execute(q))
+
+    def keys(self):
+        return list(self.iterkeys())
+
+    def itervalues(self):
+        q = "select value from %s" % self.table
+        return (self.decode(x[0]) for x in self.conn.execute(q))
+
+    def values(self):
+        return list(self.itervalues())
+
+    def iteritems(self):
+        q = "select key, value from %s" % self.table
+        return ((self.decode(x[0]), self.decode(x[1])) for x in self.conn.execute(q))
+
+    def items(self):
+        return list(self.iteritems())
+
+    def encode(self, obj):
+        return obj
+
+    def decode(self, text):
+        return text
+
+
+class PickleSqliteDict(SqliteDict):
+
+    def encode(self, obj):
+        return buffer(cPickle.dumps(obj, protocol=2))
+
+    def decode(self, text):
+        return cPickle.loads(str(text))
+
+
+class JsonSqliteDict(SqliteDict):
+
+    def encode(self, obj):
+        return json.dumps(obj)
+
+    def decode(self, text):
+        return json.loads(text)
+
+
+
+class SqlitePriorityQueue(object):
+    """SQLite priority queue. It relies on SQLite concurrency support for
+    providing atomic inter-process operations.
+    """
+
+    def __init__(self, database=':memory:', table="queue"):
+        self.database = database
+        self.table = table
+        self.conn = sqlite3.connect(database)
+        q = "create table if not exists %s (id integer primary key, " \
+            "priority real key, message blob)" % table
+        self.conn.execute(q)
+
+    def put(self, message, priority=0.0):
+        args = (priority, self.encode(message))
+        q = "insert into %s (priority, message) values (?,?)" % self.table
+        self.conn.execute(q, args)
+        self.conn.commit()
+
+    def pop(self):
+        q = "select id, message from %s order by priority desc limit 1" \
+            % self.table
+        idmsg = self.conn.execute(q).fetchone()
+        if idmsg is None:
+            return
+        id, msg = idmsg
+        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.pop()
+        self.conn.commit()
+        return self.decode(msg)
+
+    def clear(self):
+        self.conn.execute("delete from %s" % self.table)
+        self.conn.commit()
+
+    def __len__(self):
+        q = "select count(*) from %s" % self.table
+        return self.conn.execute(q).fetchone()[0]
+
+    def __iter__(self):
+        q = "select message, priority from %s order by priority desc" % \
+            self.table
+        return ((self.decode(x), y) for x, y in self.conn.execute(q))
+
+    def encode(self, obj):
+        return obj
+
+    def decode(self, text):
+        return text
+
+
+class PickleSqlitePriorityQueue(SqlitePriorityQueue):
+
+    def encode(self, obj):
+        return buffer(cPickle.dumps(obj, protocol=2))
+
+    def decode(self, text):
+        return cPickle.loads(str(text))
+
+
+class JsonSqlitePriorityQueue(SqlitePriorityQueue):
+
+    def encode(self, obj):
+        return json.dumps(obj)
+
+    def decode(self, text):
+        return json.loads(text)
+
+