Merge pull request #4434 from BroodingKangaroo/ISSUE-4250-add_batch_deliveries

Feed exports: add batch deliveries

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 <python:old-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

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

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',

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"))

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'<?xml version="1.0" encoding="utf-8"?>\n<items></items>'),
+            ('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': [
+                (
+                    '<?xml version="1.0" encoding="latin-1"?>\n'
+                    '<items>\n  <item>\n    <foo>FOO</foo>\n  </item>\n</items>'
+                ).encode('latin-1'),
+                (
+                    '<?xml version="1.0" encoding="latin-1"?>\n'
+                    '<items>\n  <item>\n    <foo>FOO1</foo>\n  </item>\n</items>'
+                ).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)

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,
         })
 
 

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 =