diff --git a/docs/topics/feed-exports.rst b/docs/topics/feed-exports.rst index 0dc73c513..dd4eb3c61 100644 --- a/docs/topics/feed-exports.rst +++ b/docs/topics/feed-exports.rst @@ -241,6 +241,7 @@ These are the settings used for configuring the feed exports: * :setting:`FEED_STORAGE_FTP_ACTIVE` * :setting:`FEED_STORAGE_S3_ACL` * :setting:`FEED_EXPORTERS` + * :setting:`FEED_EXPORT_BATCH_ITEM_COUNT` .. currentmodule:: scrapy.extensions.feedexport @@ -292,6 +293,7 @@ as a fallback value if that key is not provided for a specific feed definition. * ``fields``: falls back to :setting:`FEED_EXPORT_FIELDS` * ``indent``: falls back to :setting:`FEED_EXPORT_INDENT` * ``store_empty``: falls back to :setting:`FEED_STORE_EMPTY` +* ``batch_item_count``: falls back to :setting:`FEED_EXPORT_BATCH_ITEM_COUNT` .. setting:: FEED_EXPORT_ENCODING @@ -446,6 +448,51 @@ format in :setting:`FEED_EXPORTERS`. E.g., to disable the built-in CSV exporter 'csv': None, } + +.. setting:: FEED_EXPORT_BATCH_ITEM_COUNT + +FEED_EXPORT_BATCH_ITEM_COUNT +----------------------------- + +Default: ``0`` + +If assigned an integer number higher than ``0``, Scrapy generates multiple output files +storing up to the specified number of items in each output file. + +When generating multiple output files, you must use at least one of the following +placeholders in the feed URI to indicate how the different output file names are +generated: + +* ``%(batch_time)s`` - gets replaced by a timestamp when the feed is being created + (e.g. ``2020-03-28T14-45-08.237134``) + +* ``%(batch_id)d`` - gets replaced by the sequence number of the batch. + + Use :ref:`printf-style string formatting ` to + alter the number format. For example, to make the batch ID a 5-digit + number by introducing leading zeroes as needed, use ``%(batch_id)05d`` + (e.g. ``3`` becomes ``00003``, ``123`` becomes ``00123``). + +For instance, if your settings include:: + + FEED_EXPORT_BATCH_ITEM_COUNT = 100 + +And your :command:`crawl` command line is:: + + scrapy crawl spidername -o "dirname/%(batch_id)d-filename%(batch_time)s.json" + +The command line above can generate a directory tree like:: + +->projectname +-->dirname +--->1-filename2020-03-28T14-45-08.237134.json +--->2-filename2020-03-28T14-45-09.148903.json +--->3-filename2020-03-28T14-45-10.046092.json + +Where the first and second files contain exactly 100 items. The last one contains +100 items or fewer. + + .. _URIs: https://en.wikipedia.org/wiki/Uniform_Resource_Identifier .. _Amazon S3: https://aws.amazon.com/s3/ .. _botocore: https://github.com/boto/botocore diff --git a/scrapy/extensions/feedexport.py b/scrapy/extensions/feedexport.py index 72a34ae0d..e52f36a78 100644 --- a/scrapy/extensions/feedexport.py +++ b/scrapy/extensions/feedexport.py @@ -6,6 +6,7 @@ See documentation in docs/topics/feed-exports.rst import logging import os +import re import sys import warnings from datetime import datetime @@ -206,14 +207,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, uri_template): 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.uri_template = uri_template + self.uri = uri # flags self.itemcount = 0 self._exporting = False @@ -270,63 +273,112 @@ class FeedExporter: for uri, feed in self.feeds.items(): if not self._storage_supported(uri): raise NotConfigured + if not self._settings_are_valid(): + 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']) + self.slots.append(self._start_new_batch( + batch_id=1, + uri=uri % uri_params, + feed=feed, + spider=spider, + uri_template=uri, + )) def close_spider(self, spider): deferred_list = [] for slot in self.slots: - if not slot.itemcount and not slot.store_empty: - # We need to call slot.storage.store nonetheless to get the file - # properly closed. - d = defer.maybeDeferred(slot.storage.store, slot.file) - deferred_list.append(d) - continue - slot.finish_exporting() - logfmt = "%s %%(format)s feed (%%(itemcount)d items) in: %%(uri)s" - log_args = {'format': slot.format, - 'itemcount': slot.itemcount, - 'uri': slot.uri} - d = defer.maybeDeferred(slot.storage.store, slot.file) - - # Use `largs=log_args` to copy log_args into function's scope - # instead of using `log_args` from the outer scope - d.addCallback( - lambda _, largs=log_args: logger.info( - logfmt % "Stored", largs, extra={'spider': spider} - ) - ) - d.addErrback( - lambda f, largs=log_args: logger.error( - logfmt % "Error storing", largs, - exc_info=failure_to_exc_info(f), extra={'spider': spider} - ) - ) + d = self._close_slot(slot, spider) deferred_list.append(d) return defer.DeferredList(deferred_list) if deferred_list else None + def _close_slot(self, slot, spider): + if not slot.itemcount and not slot.store_empty: + # We need to call slot.storage.store nonetheless to get the file + # properly closed. + return defer.maybeDeferred(slot.storage.store, slot.file) + slot.finish_exporting() + logfmt = "%s %%(format)s feed (%%(itemcount)d items) in: %%(uri)s" + log_args = {'format': slot.format, + 'itemcount': slot.itemcount, + 'uri': slot.uri} + d = defer.maybeDeferred(slot.storage.store, slot.file) + + # Use `largs=log_args` to copy log_args into function's scope + # instead of using `log_args` from the outer scope + d.addCallback( + lambda _, largs=log_args: logger.info( + logfmt % "Stored", largs, extra={'spider': spider} + ) + ) + d.addErrback( + lambda f, largs=log_args: logger.error( + logfmt % "Error storing", largs, + exc_info=failure_to_exc_info(f), extra={'spider': spider} + ) + ) + return d + + def _start_new_batch(self, batch_id, uri, feed, spider, uri_template): + """ + Redirect the output data stream to a new file. + Execute multiple times if FEED_EXPORT_BATCH_ITEM_COUNT setting or FEEDS.batch_item_count is specified + :param batch_id: sequence number of current batch + :param uri: uri of the new batch to start + :param feed: dict with parameters of feed + :param spider: user spider + :param uri_template: template of uri which contains %(batch_time)s or %(batch_id)d to create new uri + """ + 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=batch_id, + uri_template=uri_template, + ) + if slot.store_empty: + slot.start_exporting() + return slot + def item_scraped(self, item, spider): + slots = [] for slot in self.slots: slot.start_exporting() slot.exporter.export_item(item) slot.itemcount += 1 + # create new slot for each slot with itemcount == FEED_EXPORT_BATCH_ITEM_COUNT and close the old one + if ( + self.feeds[slot.uri_template]['batch_item_count'] + and slot.itemcount >= self.feeds[slot.uri_template]['batch_item_count'] + ): + uri_params = self._get_uri_params(spider, self.feeds[slot.uri_template]['uri_params'], slot) + self._close_slot(slot, spider) + slots.append(self._start_new_batch( + batch_id=slot.batch_id + 1, + uri=slot.uri_template % uri_params, + feed=self.feeds[slot.uri_template], + spider=spider, + uri_template=slot.uri_template, + )) + else: + slots.append(slot) + self.slots = slots def _load_components(self, setting_prefix): conf = without_none_values(self.settings.getwithbase(setting_prefix)) @@ -343,6 +395,22 @@ class FeedExporter: return True logger.error("Unknown feed format: %(format)s", {'format': format}) + def _settings_are_valid(self): + """ + If FEED_EXPORT_BATCH_ITEM_COUNT setting or FEEDS.batch_item_count is specified uri has to contain + %(batch_time)s or %(batch_id)d to distinguish different files of partial output + """ + for uri_template, values in self.feeds.items(): + if values['batch_item_count'] and not re.search(r'%\(batch_time\)s|%\(batch_id\)', uri_template): + logger.error( + '%(batch_time)s or %(batch_id)d must be in the feed URI ({}) if FEED_EXPORT_BATCH_ITEM_COUNT ' + 'setting or FEEDS.batch_item_count is specified and greater than 0. For more info see: ' + 'https://docs.scrapy.org/en/latest/topics/feed-exports.html#feed-export-batch-item-count' + ''.format(uri_template) + ) + return False + return True + def _storage_supported(self, uri): scheme = urlparse(uri).scheme if scheme in self.storages: @@ -368,12 +436,14 @@ 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=None): params = {} for k in dir(spider): params[k] = getattr(spider, k) - ts = datetime.utcnow().replace(microsecond=0).isoformat().replace(':', '-') - params['time'] = ts + utc_now = datetime.utcnow() + params['time'] = utc_now.replace(microsecond=0).isoformat().replace(':', '-') + params['batch_time'] = utc_now.isoformat().replace(':', '-') + params['batch_id'] = slot.batch_id + 1 if slot is not None else 1 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 2fd1abbc1..896afa995 100644 --- a/scrapy/settings/default_settings.py +++ b/scrapy/settings/default_settings.py @@ -147,6 +147,7 @@ FEED_STORAGES_BASE = { 's3': 'scrapy.extensions.feedexport.S3FeedStorage', 'stdout': 'scrapy.extensions.feedexport.StdoutFeedStorage', } +FEED_EXPORT_BATCH_ITEM_COUNT = 0 FEED_EXPORTERS = {} FEED_EXPORTERS_BASE = { 'json': 'scrapy.exporters.JsonItemExporter', diff --git a/scrapy/utils/conf.py b/scrapy/utils/conf.py index 728bb5f1b..a83076c47 100644 --- a/scrapy/utils/conf.py +++ b/scrapy/utils/conf.py @@ -115,6 +115,7 @@ def get_sources(use_closest=True): def feed_complete_default_values_from_settings(feed, settings): out = feed.copy() + out.setdefault("batch_item_count", settings.getint('FEED_EXPORT_BATCH_ITEM_COUNT')) out.setdefault("encoding", settings["FEED_EXPORT_ENCODING"]) out.setdefault("fields", settings.getlist("FEED_EXPORT_FIELDS") or None) out.setdefault("store_empty", settings.getbool("FEED_STORE_EMPTY")) diff --git a/tests/test_feedexport.py b/tests/test_feedexport.py index 37384081a..c49b2e92f 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 logging import getLogger from pathlib import Path @@ -24,9 +26,11 @@ 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, + FeedExporter, FileFeedStorage, FTPFeedStorage, GCSFeedStorage, @@ -91,6 +95,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 @@ -144,6 +149,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 @@ -502,23 +508,84 @@ class LogOnStoreFileStorage: file.close() -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) + 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 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): @@ -559,35 +626,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 {} @@ -653,18 +691,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 {} @@ -693,15 +719,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 @@ -725,7 +742,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): @@ -745,7 +762,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_no_items_multiple_feeds(self): @@ -1163,3 +1180,376 @@ class FeedExportTest(unittest.TestCase): print(log) for fmt in ['json', 'xml', 'csv']: self.assertIn('Error storing %s feed (2 items)' % fmt, str(log)) + + +class BatchDeliveriesTest(FeedExportTestBase): + __test__ = True + _file_mark = '_%(batch_time)s_#%(batch_id)02d_' + + @defer.inlineCallbacks + def run_and_export(self, spider_cls, settings): + """ Run spider with specified settings; return exported data. """ + + def build_url(path): + if path[0] != '/': + path = '/' + path + return urljoin('file:', path) + + FEEDS = settings.get('FEEDS') or {} + settings['FEEDS'] = { + build_url(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.getint('FEED_EXPORT_BATCH_ITEM_COUNT') + 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.getint('FEED_EXPORT_BATCH_ITEM_COUNT') + 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.getint('FEED_EXPORT_BATCH_ITEM_COUNT') + 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.getint('FEED_EXPORT_BATCH_ITEM_COUNT') + 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.getint('FEED_EXPORT_BATCH_ITEM_COUNT') + 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.getint('FEED_EXPORT_BATCH_ITEM_COUNT') + 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): + """ Test partial deliveries in all supported formats """ + 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_EXPORT_BATCH_ITEM_COUNT': 2 + } + header = self.MyItem.fields.keys() + yield self.assertExported(items, header, rows, settings=Settings(settings)) + + def test_wrong_path(self): + """ If path is without %(batch_time)s and %(batch_id) an exception must be raised """ + settings = { + 'FEEDS': { + self._random_temp_filename(): {'format': 'xml'}, + }, + 'FEED_EXPORT_BATCH_ITEM_COUNT': 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_EXPORT_BATCH_ITEM_COUNT': 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_EXPORT_BATCH_ITEM_COUNT': 1, + } + data = yield self.exported_no_data(settings) + data = dict(data) + self.assertEqual(expctd, data[fmt][0]) + + @defer.inlineCallbacks + def test_export_multiple_configs(self): + items = [dict({'foo': u'FOO', 'bar': u'BAR'}), dict({'foo': u'FOO1', 'bar': u'BAR1'})] + + formats = { + 'json': ['[\n{"bar": "BAR"}\n]'.encode('utf-8'), + '[\n{"bar": "BAR1"}\n]'.encode('utf-8')], + 'xml': [ + ( + '\n' + '\n \n FOO\n \n' + ).encode('latin-1'), + ( + '\n' + '\n \n FOO1\n \n' + ).encode('latin-1') + ], + 'csv': ['foo,bar\r\nFOO,BAR\r\n'.encode('utf-8'), + 'foo,bar\r\nFOO1,BAR1\r\n'.encode('utf-8')], + } + + settings = { + 'FEEDS': { + os.path.join(self._random_temp_filename(), 'json', self._file_mark): { + 'format': 'json', + 'indent': 0, + 'fields': ['bar'], + 'encoding': 'utf-8', + }, + os.path.join(self._random_temp_filename(), 'xml', self._file_mark): { + 'format': 'xml', + 'indent': 2, + 'fields': ['foo'], + 'encoding': 'latin-1', + }, + os.path.join(self._random_temp_filename(), 'csv', self._file_mark): { + 'format': 'csv', + 'indent': None, + 'fields': ['foo', 'bar'], + 'encoding': 'utf-8', + }, + }, + 'FEED_EXPORT_BATCH_ITEM_COUNT': 1, + } + data = yield self.exported_data(items, settings) + for fmt, expected in formats.items(): + for expected_batch, got_batch in zip(expected, data[fmt]): + self.assertEqual(expected_batch, got_batch) + + @defer.inlineCallbacks + def test_batch_item_count_feeds_setting(self): + items = [dict({'foo': u'FOO'}), dict({'foo': u'FOO1'})] + formats = { + 'json': ['[{"foo": "FOO"}]'.encode('utf-8'), + '[{"foo": "FOO1"}]'.encode('utf-8')], + } + settings = { + 'FEEDS': { + os.path.join(self._random_temp_filename(), 'json', self._file_mark): { + 'format': 'json', + 'indent': None, + 'encoding': 'utf-8', + 'batch_item_count': 1, + }, + }, + } + data = yield self.exported_data(items, settings) + for fmt, expected in formats.items(): + for expected_batch, got_batch in zip(expected, data[fmt]): + self.assertEqual(expected_batch, got_batch) + + @defer.inlineCallbacks + def test_batch_path_differ(self): + """ + Test that the name of all batch files differ from each other. + So %(batch_time)s replaced with the current date. + """ + items = [ + self.MyItem({'foo': 'bar1', 'egg': 'spam1'}), + self.MyItem({'foo': 'bar2', 'egg': 'spam2', 'baz': 'quux2'}), + self.MyItem({'foo': 'bar3', 'baz': 'quux3'}), + ] + settings = { + 'FEEDS': { + os.path.join(self._random_temp_filename(), '%(batch_time)s'): { + 'format': 'json', + }, + }, + 'FEED_EXPORT_BATCH_ITEM_COUNT': 1, + } + data = yield self.exported_data(items, settings) + self.assertEqual(len(items) + 1, len(data['json'])) + + @defer.inlineCallbacks + def test_s3_export(self): + """ + Test export of items into s3 bucket. + S3_TEST_BUCKET_NAME, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY must be specified in tox.ini + to perform this test: + [testenv] + setenv = + AWS_SECRET_ACCESS_KEY = ABCD + AWS_ACCESS_KEY_ID = EFGH + S3_TEST_BUCKET_NAME = IJKL + """ + try: + import boto3 + except ImportError: + raise unittest.SkipTest("S3FeedStorage requires boto3") + + assert_aws_environ() + s3_test_bucket_name = os.environ.get('S3_TEST_BUCKET_NAME') + access_key = os.environ.get('AWS_ACCESS_KEY_ID') + secret_key = os.environ.get('AWS_SECRET_ACCESS_KEY') + if not s3_test_bucket_name: + raise unittest.SkipTest("No S3 BUCKET available for testing") + + chars = [random.choice(ascii_letters + digits) for _ in range(15)] + filename = ''.join(chars) + prefix = 'tmp/{filename}'.format(filename=filename) + s3_test_file_uri = 's3://{bucket_name}/{prefix}/%(batch_time)s.json'.format( + bucket_name=s3_test_bucket_name, prefix=prefix + ) + storage = S3FeedStorage(s3_test_bucket_name, access_key, secret_key) + settings = Settings({ + 'FEEDS': { + s3_test_file_uri: { + 'format': 'json', + }, + }, + 'FEED_EXPORT_BATCH_ITEM_COUNT': 1, + }) + items = [ + self.MyItem({'foo': 'bar1', 'egg': 'spam1'}), + self.MyItem({'foo': 'bar2', 'egg': 'spam2', 'baz': 'quux2'}), + self.MyItem({'foo': 'bar3', 'baz': 'quux3'}), + ] + verifyObject(IFeedStorage, storage) + + class TestSpider(scrapy.Spider): + name = 'testspider' + + def parse(self, response): + for item in items: + yield item + + s3 = boto3.resource('s3') + my_bucket = s3.Bucket(s3_test_bucket_name) + batch_size = settings.getint('FEED_EXPORT_BATCH_ITEM_COUNT') + + with MockServer() as s: + runner = CrawlerRunner(Settings(settings)) + TestSpider.start_urls = [s.url('/')] + yield runner.crawl(TestSpider) + + for file_uri in my_bucket.objects.filter(Prefix=prefix): + content = get_s3_content_and_delete(s3_test_bucket_name, file_uri.key) + if not content and not items: + break + content = json.loads(content.decode('utf-8')) + expected_batch, items = items[:batch_size], items[batch_size:] + self.assertEqual(expected_batch, content) diff --git a/tests/test_utils_conf.py b/tests/test_utils_conf.py index e5d3ef582..f3ef36127 100644 --- a/tests/test_utils_conf.py +++ b/tests/test_utils_conf.py @@ -149,6 +149,7 @@ class FeedExportConfigTestCase(unittest.TestCase): "FEED_EXPORT_INDENT": 42, "FEED_STORE_EMPTY": True, "FEED_URI_PARAMS": (1, 2, 3, 4), + "FEED_EXPORT_BATCH_ITEM_COUNT": 2, }) new_feed = feed_complete_default_values_from_settings(feed, settings) self.assertEqual(new_feed, { @@ -157,6 +158,7 @@ class FeedExportConfigTestCase(unittest.TestCase): "indent": 42, "store_empty": True, "uri_params": (1, 2, 3, 4), + "batch_item_count": 2, }) def test_feed_complete_default_values_from_settings_non_empty(self): @@ -169,6 +171,7 @@ class FeedExportConfigTestCase(unittest.TestCase): "FEED_EXPORT_FIELDS": ["f1", "f2", "f3"], "FEED_EXPORT_INDENT": 42, "FEED_STORE_EMPTY": True, + "FEED_EXPORT_BATCH_ITEM_COUNT": 2, }) new_feed = feed_complete_default_values_from_settings(feed, settings) self.assertEqual(new_feed, { @@ -177,6 +180,7 @@ class FeedExportConfigTestCase(unittest.TestCase): "indent": 42, "store_empty": True, "uri_params": None, + "batch_item_count": 2, }) diff --git a/tox.ini b/tox.ini index 4557c63e3..11882c03f 100644 --- a/tox.ini +++ b/tox.ini @@ -12,6 +12,7 @@ deps = -ctests/constraints.txt -rtests/requirements-py3.txt # Extras + boto3>=1.13.0 botocore>=1.3.23 Pillow>=3.4.2 passenv =