From 2ff5a83b7a16c74b55d8765001ce8832d8d5c634 Mon Sep 17 00:00:00 2001 From: Pablo Hoffman Date: Thu, 19 Aug 2010 02:55:52 -0300 Subject: [PATCH] Added persistent execution queue (based on SQLite), and a new 'queue' command to control it. Closes #198 --- docs/topics/commands.rst | 38 +++++++- scrapy/commands/queue.py | 46 +++++++++ scrapy/conf/default_settings.py | 3 +- scrapy/contrib/queue/__init__.py | 20 ++++ scrapy/tests/test_utils_sqlite.py | 141 +++++++++++++++++++++++++++ scrapy/utils/sqlite.py | 155 ++++++++++++++++++++++++++++++ 6 files changed, 401 insertions(+), 2 deletions(-) create mode 100644 scrapy/commands/queue.py create mode 100644 scrapy/tests/test_utils_sqlite.py create mode 100644 scrapy/utils/sqlite.py diff --git a/docs/topics/commands.rst b/docs/topics/commands.rst index 6a5c4191f..56f625f34 100644 --- a/docs/topics/commands.rst +++ b/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 `` | ++-------------------+----------------------------------------------+ +| 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 diff --git a/scrapy/commands/queue.py b/scrapy/commands/queue.py new file mode 100644 index 000000000..f1d43b0a5 --- /dev/null +++ b/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] " + + 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 diff --git a/scrapy/conf/default_settings.py b/scrapy/conf/default_settings.py index 28c0e10ae..b0b3cd1f0 100644 --- a/scrapy/conf/default_settings.py +++ b/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' diff --git a/scrapy/contrib/queue/__init__.py b/scrapy/contrib/queue/__init__.py index e69de29bb..fe39f07ce 100644 --- a/scrapy/contrib/queue/__init__.py +++ b/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 diff --git a/scrapy/tests/test_utils_sqlite.py b/scrapy/tests/test_utils_sqlite.py new file mode 100644 index 000000000..f42240e01 --- /dev/null +++ b/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) diff --git a/scrapy/utils/sqlite.py b/scrapy/utils/sqlite.py new file mode 100644 index 000000000..5451550dc --- /dev/null +++ b/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) + +