diff --git a/scrapy/extensions/feedexport.py b/scrapy/extensions/feedexport.py index 998d2a5d1..06ea6c5b2 100644 --- a/scrapy/extensions/feedexport.py +++ b/scrapy/extensions/feedexport.py @@ -25,7 +25,6 @@ from scrapy.utils.log import failure_to_exc_info from scrapy.utils.misc import create_instance, load_object from scrapy.utils.python import without_none_values - logger = logging.getLogger(__name__) @@ -180,14 +179,16 @@ class FTPFeedStorage(BlockingFeedStorage): class _FeedSlot: - def __init__(self, file, exporter, storage, uri, format, store_empty): + def __init__(self, file, exporter, storage, uri, format, store_empty, batch_id, template_uri): self.file = file self.exporter = exporter self.storage = storage # feed params - self.uri = uri + self.batch_id = batch_id self.format = format self.store_empty = store_empty + self.template_uri = template_uri + self.uri = uri # flags self.itemcount = 0 self._exporting = False @@ -241,28 +242,25 @@ class FeedExporter: self.storages = self._load_components('FEED_STORAGES') self.exporters = self._load_components('FEED_EXPORTERS') + self.storage_batch_size = self.settings.get('FEED_STORAGE_BATCH_SIZE', None) for uri, feed in self.feeds.items(): if not self._storage_supported(uri): raise NotConfigured + if not self._batch_deliveries_supported(uri): + raise NotConfigured if not self._exporter_supported(feed['format']): raise NotConfigured def open_spider(self, spider): for uri, feed in self.feeds.items(): - uri = uri % self._get_uri_params(spider, feed['uri_params']) - storage = self._get_storage(uri) - file = storage.open(spider) - exporter = self._get_exporter( - file=file, - format=feed['format'], - fields_to_export=feed['fields'], - encoding=feed['encoding'], - indent=feed['indent'], - ) - slot = _FeedSlot(file, exporter, storage, uri, feed['format'], feed['store_empty']) - self.slots.append(slot) - if slot.store_empty: - slot.start_exporting() + uri_params = self._get_uri_params(spider, feed['uri_params'], None) + self.slots.append(self._start_new_batch( + previous_batch_slot=None, + uri=uri % uri_params, + feed=feed, + spider=spider, + template_uri=uri, + )) def close_spider(self, spider): deferred_list = [] @@ -285,11 +283,65 @@ class FeedExporter: deferred_list.append(d) return defer.DeferredList(deferred_list) if deferred_list else None + def _start_new_batch(self, previous_batch_slot, uri, feed, spider, template_uri): + """ + Redirect the output data stream to a new file. + Execute multiple times if 'FEED_STORAGE_BATCH' setting is specified. + :param previous_batch_slot: slot of previous batch. We need to call slot.storage.store + to get the file properly closed. + :param uri: uri of the new batch to start + :param feed: dict with parameters of feed + :param spider: user spider + :param template_uri: template uri which contains %(time)s or %(batch_id)s to create new uri + """ + if previous_batch_slot is not None: + previous_batch_id = previous_batch_slot.batch_id + previous_batch_slot.exporter.finish_exporting() + previous_batch_slot.storage.store(previous_batch_slot.file) + else: + previous_batch_id = 0 + + storage = self._get_storage(uri) + file = storage.open(spider) + exporter = self._get_exporter( + file=file, + format=feed['format'], + fields_to_export=feed['fields'], + encoding=feed['encoding'], + indent=feed['indent'], + ) + slot = _FeedSlot( + file=file, + exporter=exporter, + storage=storage, + uri=uri, + format=feed['format'], + store_empty=feed['store_empty'], + batch_id=previous_batch_id + 1, + template_uri=template_uri, + ) + if slot.store_empty: + slot.start_exporting() + return slot + def item_scraped(self, item, spider): - for slot in self.slots: + slots = [] + for idx, slot in enumerate(self.slots): slot.start_exporting() slot.exporter.export_item(item) slot.itemcount += 1 + if self.storage_batch_size and slot.itemcount % self.storage_batch_size == 0: + uri_params = self._get_uri_params(spider, self.feeds[slot.template_uri]['uri_params'], slot) + slots.append(self._start_new_batch( + previous_batch_slot=slot, + uri=slot.template_uri % uri_params, + feed=self.feeds[slot.template_uri], + spider=spider, + template_uri=slot.template_uri, + )) + self.slots[idx] = None + self.slots = [slot for slot in self.slots if slot is not None] + self.slots.extend(slots) def _load_components(self, setting_prefix): conf = without_none_values(self.settings.getwithbase(setting_prefix)) @@ -306,6 +358,16 @@ class FeedExporter: return True logger.error("Unknown feed format: %(format)s", {'format': format}) + def _batch_deliveries_supported(self, uri): + """ + If FEED_STORAGE_BATCH_SIZE setting is specified uri has to contain %(time)s or %(batch_id)s + to distinguish different files of partial output + """ + if self.storage_batch_size is None or '%(time)s' in uri or '%(batch_id)s' in uri: + return True + logger.warning('%(time)s or %(batch_id)s must be in uri if FEED_STORAGE_BATCH_SIZE setting is specified') + return False + def _storage_supported(self, uri): scheme = urlparse(uri).scheme if scheme in self.storages: @@ -331,12 +393,12 @@ class FeedExporter: def _get_storage(self, uri): return self._get_instance(self.storages[urlparse(uri).scheme], uri) - def _get_uri_params(self, spider, uri_params): + def _get_uri_params(self, spider, uri_params, slot): params = {} for k in dir(spider): params[k] = getattr(spider, k) - ts = datetime.utcnow().replace(microsecond=0).isoformat().replace(':', '-') - params['time'] = ts + params['batch_id'] = slot.batch_id + 1 if slot is not None else 1 + params['time'] = datetime.utcnow().isoformat().replace(':', '-') uripar_function = load_object(uri_params) if uri_params else lambda x, y: None uripar_function(params, spider) return params diff --git a/scrapy/settings/default_settings.py b/scrapy/settings/default_settings.py index 077317c81..c3463a505 100644 --- a/scrapy/settings/default_settings.py +++ b/scrapy/settings/default_settings.py @@ -146,6 +146,7 @@ FEED_STORAGES_BASE = { 's3': 'scrapy.extensions.feedexport.S3FeedStorage', 'ftp': 'scrapy.extensions.feedexport.FTPFeedStorage', } +FEED_STORAGE_BATCH_SIZE = None FEED_EXPORTERS = {} FEED_EXPORTERS_BASE = { 'json': 'scrapy.exporters.JsonItemExporter', diff --git a/tests/test_feedexport.py b/tests/test_feedexport.py index e02b0b840..60e19d1df 100644 --- a/tests/test_feedexport.py +++ b/tests/test_feedexport.py @@ -6,6 +6,8 @@ import shutil import string import tempfile import warnings +from abc import ABC, abstractmethod +from collections import defaultdict from io import BytesIO from pathlib import Path from string import ascii_letters, digits @@ -21,8 +23,9 @@ from zope.interface.verify import verifyObject import scrapy from scrapy.crawler import CrawlerRunner +from scrapy.exceptions import NotConfigured from scrapy.exporters import CsvItemExporter -from scrapy.extensions.feedexport import (BlockingFeedStorage, FileFeedStorage, FTPFeedStorage, +from scrapy.extensions.feedexport import (BlockingFeedStorage, FeedExporter, FileFeedStorage, FTPFeedStorage, IFeedStorage, S3FeedStorage, StdoutFeedStorage) from scrapy.settings import Settings from scrapy.utils.python import to_unicode @@ -76,6 +79,7 @@ class FTPFeedStorageTest(unittest.TestCase): def get_test_spider(self, settings=None): class TestSpider(scrapy.Spider): name = 'test_spider' + crawler = get_crawler(settings_dict=settings) spider = TestSpider.from_crawler(crawler) return spider @@ -129,6 +133,7 @@ class BlockingFeedStorageTest(unittest.TestCase): def get_test_spider(self, settings=None): class TestSpider(scrapy.Spider): name = 'test_spider' + crawler = get_crawler(settings_dict=settings) spider = TestSpider.from_crawler(crawler) return spider @@ -390,23 +395,84 @@ class FromCrawlerFileFeedStorage(FileFeedStorage, FromCrawlerMixin): pass -class FeedExportTest(unittest.TestCase): +class FeedExportTestBase(ABC, unittest.TestCase): + __test__ = False class MyItem(scrapy.Item): foo = scrapy.Field() egg = scrapy.Field() baz = scrapy.Field() + def _random_temp_filename(self, inter_dir=''): + chars = [random.choice(ascii_letters + digits) for _ in range(15)] + filename = ''.join(chars) + return os.path.join(self.temp_dir, inter_dir, filename) + def setUp(self): self.temp_dir = tempfile.mkdtemp() def tearDown(self): shutil.rmtree(self.temp_dir, ignore_errors=True) - def _random_temp_filename(self): - chars = [random.choice(ascii_letters + digits) for _ in range(15)] - filename = ''.join(chars) - return os.path.join(self.temp_dir, filename) + @defer.inlineCallbacks + def exported_data(self, items, settings): + """ + Return exported data which a spider yielding ``items`` would return. + """ + + class TestSpider(scrapy.Spider): + name = 'testspider' + + def parse(self, response): + for item in items: + yield item + + data = yield self.run_and_export(TestSpider, settings) + defer.returnValue(data) + + @defer.inlineCallbacks + def exported_no_data(self, settings): + """ + Return exported data which a spider yielding no ``items`` would return. + """ + + class TestSpider(scrapy.Spider): + name = 'testspider' + + def parse(self, response): + pass + + data = yield self.run_and_export(TestSpider, settings) + defer.returnValue(data) + + @defer.inlineCallbacks + def assertExported(self, items, header, rows, settings=None, ordered=True): + yield self.assertExportedCsv(items, header, rows, settings, ordered) + yield self.assertExportedJsonLines(items, rows, settings) + yield self.assertExportedXml(items, rows, settings) + yield self.assertExportedPickle(items, rows, settings) + yield self.assertExportedMarshal(items, rows, settings) + yield self.assertExportedMultiple(items, rows, settings) + + @abstractmethod + def run_and_export(self, spider_cls, settings): + pass + + def _load_until_eof(self, data, load_func): + result = [] + with tempfile.TemporaryFile() as temp: + temp.write(data) + temp.seek(0) + while True: + try: + result.append(load_func(temp)) + except EOFError: + break + return result + + +class FeedExportTest(FeedExportTestBase): + __test__ = True @defer.inlineCallbacks def run_and_export(self, spider_cls, settings): @@ -417,7 +483,6 @@ class FeedExportTest(unittest.TestCase): urljoin('file:', pathname2url(str(file_path))): feed for file_path, feed in FEEDS.items() } - content = {} try: with MockServer() as s: @@ -435,35 +500,6 @@ class FeedExportTest(unittest.TestCase): return content - @defer.inlineCallbacks - def exported_data(self, items, settings): - """ - Return exported data which a spider yielding ``items`` would return. - """ - class TestSpider(scrapy.Spider): - name = 'testspider' - - def parse(self, response): - for item in items: - yield item - - data = yield self.run_and_export(TestSpider, settings) - return data - - @defer.inlineCallbacks - def exported_no_data(self, settings): - """ - Return exported data which a spider yielding no ``items`` would return. - """ - class TestSpider(scrapy.Spider): - name = 'testspider' - - def parse(self, response): - pass - - data = yield self.run_and_export(TestSpider, settings) - return data - @defer.inlineCallbacks def assertExportedCsv(self, items, header, rows, settings=None, ordered=True): settings = settings or {} @@ -529,18 +565,6 @@ class FeedExportTest(unittest.TestCase): json_rows = json.loads(to_unicode(data['json'])) self.assertEqual(rows, json_rows) - def _load_until_eof(self, data, load_func): - result = [] - with tempfile.TemporaryFile() as temp: - temp.write(data) - temp.seek(0) - while True: - try: - result.append(load_func(temp)) - except EOFError: - break - return result - @defer.inlineCallbacks def assertExportedPickle(self, items, rows, settings=None): settings = settings or {} @@ -569,15 +593,6 @@ class FeedExportTest(unittest.TestCase): result = self._load_until_eof(data['marshal'], load_func=marshal.load) self.assertEqual(expected, result) - @defer.inlineCallbacks - def assertExported(self, items, header, rows, settings=None, ordered=True): - yield self.assertExportedCsv(items, header, rows, settings, ordered) - yield self.assertExportedJsonLines(items, rows, settings) - yield self.assertExportedXml(items, rows, settings) - yield self.assertExportedPickle(items, rows, settings) - yield self.assertExportedMarshal(items, rows, settings) - yield self.assertExportedMultiple(items, rows, settings) - @defer.inlineCallbacks def test_export_items(self): # feed exporters use field names from Item @@ -601,7 +616,7 @@ class FeedExportTest(unittest.TestCase): }, } data = yield self.exported_no_data(settings) - self.assertEqual(data[fmt], b'') + self.assertEqual(b'', data[fmt]) @defer.inlineCallbacks def test_export_no_items_store_empty(self): @@ -621,7 +636,7 @@ class FeedExportTest(unittest.TestCase): 'FEED_EXPORT_INDENT': None, } data = yield self.exported_no_data(settings) - self.assertEqual(data[fmt], expctd) + self.assertEqual(expctd, data[fmt]) @defer.inlineCallbacks def test_export_multiple_item_classes(self): @@ -720,7 +735,8 @@ class FeedExportTest(unittest.TestCase): formats = { 'json': u'[{"foo": "Test\\u00d6"}]'.encode('utf-8'), 'jsonlines': u'{"foo": "Test\\u00d6"}\n'.encode('utf-8'), - 'xml': u'\nTest\xd6'.encode('utf-8'), + 'xml': u'\nTest\xd6'.encode( + 'utf-8'), 'csv': u'foo\r\nTest\xd6\r\n'.encode('utf-8'), } @@ -737,7 +753,8 @@ class FeedExportTest(unittest.TestCase): formats = { 'json': u'[{"foo": "Test\xd6"}]'.encode('latin-1'), 'jsonlines': u'{"foo": "Test\xd6"}\n'.encode('latin-1'), - 'xml': u'\nTest\xd6'.encode('latin-1'), + 'xml': u'\nTest\xd6'.encode( + 'latin-1'), 'csv': u'foo\r\nTest\xd6\r\n'.encode('latin-1'), } @@ -758,7 +775,8 @@ class FeedExportTest(unittest.TestCase): formats = { 'json': u'[\n{"bar": "BAR"}\n]'.encode('utf-8'), - 'xml': u'\n\n \n FOO\n \n'.encode('latin-1'), + 'xml': u'\n\n \n FOO\n \n'.encode( + 'latin-1'), 'csv': u'bar,foo\r\nBAR,FOO\r\n'.encode('utf-8'), } @@ -970,3 +988,207 @@ class FeedExportTest(unittest.TestCase): } data = yield self.exported_no_data(settings) self.assertEqual(data['csv'], b'') + + +class PartialDeliveriesTest(FeedExportTestBase): + __test__ = True + _file_mark = '_%(time)s_#%(batch_id)s_' + + @defer.inlineCallbacks + def run_and_export(self, spider_cls, settings): + """ Run spider with specified settings; return exported data. """ + + FEEDS = settings.get('FEEDS') or {} + settings['FEEDS'] = { + urljoin('file:', file_path): feed + for file_path, feed in FEEDS.items() + } + content = defaultdict(list) + try: + with MockServer() as s: + runner = CrawlerRunner(Settings(settings)) + spider_cls.start_urls = [s.url('/')] + yield runner.crawl(spider_cls) + + for path, feed in FEEDS.items(): + dir_name = os.path.dirname(path) + for file in sorted(os.listdir(dir_name)): + with open(os.path.join(dir_name, file), 'rb') as f: + data = f.read() + content[feed['format']].append(data) + finally: + self.tearDown() + defer.returnValue(content) + + @defer.inlineCallbacks + def assertExportedJsonLines(self, items, rows, settings=None): + settings = settings or {} + settings.update({ + 'FEEDS': { + os.path.join(self._random_temp_filename(), 'jl', self._file_mark): {'format': 'jl'}, + }, + }) + batch_size = settings['FEED_STORAGE_BATCH_SIZE'] + rows = [{k: v for k, v in row.items() if v} for row in rows] + data = yield self.exported_data(items, settings) + for batch in data['jl']: + got_batch = [json.loads(to_unicode(batch_item)) for batch_item in batch.splitlines()] + expected_batch, rows = rows[:batch_size], rows[batch_size:] + self.assertEqual(expected_batch, got_batch) + + @defer.inlineCallbacks + def assertExportedCsv(self, items, header, rows, settings=None, ordered=True): + settings = settings or {} + settings.update({ + 'FEEDS': { + os.path.join(self._random_temp_filename(), 'csv', self._file_mark): {'format': 'csv'}, + }, + }) + batch_size = settings['FEED_STORAGE_BATCH_SIZE'] + data = yield self.exported_data(items, settings) + for batch in data['csv']: + got_batch = csv.DictReader(to_unicode(batch).splitlines()) + self.assertEqual(list(header), got_batch.fieldnames) + expected_batch, rows = rows[:batch_size], rows[batch_size:] + self.assertEqual(expected_batch, list(got_batch)) + + @defer.inlineCallbacks + def assertExportedXml(self, items, rows, settings=None): + settings = settings or {} + settings.update({ + 'FEEDS': { + os.path.join(self._random_temp_filename(), 'xml', self._file_mark): {'format': 'xml'}, + }, + }) + batch_size = settings['FEED_STORAGE_BATCH_SIZE'] + rows = [{k: v for k, v in row.items() if v} for row in rows] + data = yield self.exported_data(items, settings) + for batch in data['xml']: + root = lxml.etree.fromstring(batch) + got_batch = [{e.tag: e.text for e in it} for it in root.findall('item')] + expected_batch, rows = rows[:batch_size], rows[batch_size:] + self.assertEqual(expected_batch, got_batch) + + @defer.inlineCallbacks + def assertExportedMultiple(self, items, rows, settings=None): + settings = settings or {} + settings.update({ + 'FEEDS': { + os.path.join(self._random_temp_filename(), 'xml', self._file_mark): {'format': 'xml'}, + os.path.join(self._random_temp_filename(), 'json', self._file_mark): {'format': 'json'}, + }, + }) + batch_size = settings['FEED_STORAGE_BATCH_SIZE'] + rows = [{k: v for k, v in row.items() if v} for row in rows] + data = yield self.exported_data(items, settings) + # XML + xml_rows = rows.copy() + for batch in data['xml']: + root = lxml.etree.fromstring(batch) + got_batch = [{e.tag: e.text for e in it} for it in root.findall('item')] + expected_batch, xml_rows = xml_rows[:batch_size], xml_rows[batch_size:] + self.assertEqual(expected_batch, got_batch) + # JSON + json_rows = rows.copy() + for batch in data['json']: + got_batch = json.loads(batch.decode('utf-8')) + expected_batch, json_rows = json_rows[:batch_size], json_rows[batch_size:] + self.assertEqual(expected_batch, got_batch) + + @defer.inlineCallbacks + def assertExportedPickle(self, items, rows, settings=None): + settings = settings or {} + settings.update({ + 'FEEDS': { + os.path.join(self._random_temp_filename(), 'pickle', self._file_mark): {'format': 'pickle'}, + }, + }) + batch_size = settings['FEED_STORAGE_BATCH_SIZE'] + rows = [{k: v for k, v in row.items() if v} for row in rows] + data = yield self.exported_data(items, settings) + import pickle + for batch in data['pickle']: + got_batch = self._load_until_eof(batch, load_func=pickle.load) + expected_batch, rows = rows[:batch_size], rows[batch_size:] + self.assertEqual(expected_batch, got_batch) + + @defer.inlineCallbacks + def assertExportedMarshal(self, items, rows, settings=None): + settings = settings or {} + settings.update({ + 'FEEDS': { + os.path.join(self._random_temp_filename(), 'marshal', self._file_mark): {'format': 'marshal'}, + }, + }) + batch_size = settings['FEED_STORAGE_BATCH_SIZE'] + rows = [{k: v for k, v in row.items() if v} for row in rows] + data = yield self.exported_data(items, settings) + import marshal + for batch in data['marshal']: + got_batch = self._load_until_eof(batch, load_func=marshal.load) + expected_batch, rows = rows[:batch_size], rows[batch_size:] + self.assertEqual(expected_batch, got_batch) + + @defer.inlineCallbacks + def test_export_items(self): + items = [ + self.MyItem({'foo': 'bar1', 'egg': 'spam1'}), + self.MyItem({'foo': 'bar2', 'egg': 'spam2', 'baz': 'quux2'}), + self.MyItem({'foo': 'bar3', 'baz': 'quux3'}), + ] + rows = [ + {'egg': 'spam1', 'foo': 'bar1', 'baz': ''}, + {'egg': 'spam2', 'foo': 'bar2', 'baz': 'quux2'}, + {'foo': 'bar3', 'baz': 'quux3', 'egg': ''} + ] + settings = { + 'FEED_STORAGE_BATCH_SIZE': 2 + } + header = self.MyItem.fields.keys() + yield self.assertExported(items, header, rows, settings=settings) + + def test_wrong_path(self): + """If path without %(time)s or %(batch_id)s an exception must be raised""" + settings = { + 'FEEDS': { + self._random_temp_filename(): {'format': 'xml'}, + }, + 'FEED_STORAGE_BATCH_SIZE': 1 + } + crawler = get_crawler(settings_dict=settings) + self.assertRaises(NotConfigured, FeedExporter, crawler) + + @defer.inlineCallbacks + def test_export_no_items_not_store_empty(self): + for fmt in ('json', 'jsonlines', 'xml', 'csv'): + settings = { + 'FEEDS': { + os.path.join(self._random_temp_filename(), fmt, self._file_mark): {'format': fmt}, + }, + 'FEED_STORAGE_BATCH_SIZE': 1 + } + data = yield self.exported_no_data(settings) + data = dict(data) + self.assertEqual(b'', data[fmt][0]) + + @defer.inlineCallbacks + def test_export_no_items_store_empty(self): + formats = ( + ('json', b'[]'), + ('jsonlines', b''), + ('xml', b'\n'), + ('csv', b''), + ) + + for fmt, expctd in formats: + settings = { + 'FEEDS': { + os.path.join(self._random_temp_filename(), fmt, self._file_mark): {'format': fmt}, + }, + 'FEED_STORE_EMPTY': True, + 'FEED_EXPORT_INDENT': None, + 'FEED_STORAGE_BATCH_SIZE': 1, + } + data = yield self.exported_no_data(settings) + data = dict(data) + self.assertEqual(expctd, data[fmt][0])