Repository: orangain/scrapy-s3pipeline

Issue:
Json object are not delimited and not within array using JsonLinesItemExporter
Using scrapy-s3pipeline, a but json items are not delimited or within array

see https://github.com/scrapy/scrapy/issues/4960

Discussion:

Patch:
diff --git a/README.md b/README.md
index 6b03cd5..629e810 100644
--- a/README.md
+++ b/README.md
@@ -2,9 +2,10 @@
 
 [![PyPI version](https://badge.fury.io/py/scrapy-s3pipeline.svg)](https://badge.fury.io/py/scrapy-s3pipeline) ![CI](https://github.com/orangain/scrapy-s3pipeline/workflows/CI/badge.svg)
 
-Scrapy pipeline to store items into [Amazon S3](https://aws.amazon.com/s3/) or [Google Cloud Storage (GCS)](https://cloud.google.com/storage) bucket with JSONLines format. Unlike built-in [FeedExporter](https://docs.scrapy.org/en/latest/topics/feed-exports.html#s3), the pipeline has the following features:
+Scrapy pipeline to store items into [Amazon S3](https://aws.amazon.com/s3/) or [Google Cloud Storage (GCS)](https://cloud.google.com/storage) bucket. Unlike built-in [FeedExporter](https://docs.scrapy.org/en/latest/topics/feed-exports.html#s3), the pipeline has the following features:
 
 * The pipeline upload items to S3/GCS by chunk while crawler is running.
+  * From Scrapy 2.3, built-in [FEED_EXPORT_BATCH_ITEM_COUNT](https://docs.scrapy.org/en/latest/topics/feed-exports.html#std-setting-FEED_EXPORT_BATCH_ITEM_COUNT) does almost the same thing.
 * Support GZip compression.
 
 The pipeline aims to run crawler and scraper in different processes, e.g. run crawler process with Scrapy in AWS Fargate and run scraper process with lxml in AWS Lambda.
@@ -92,6 +93,8 @@ The following replacement fields are supported in `S3PIPELINE_URL`.
 
 You can also use other spider fields, e.g. `{name}`. You can use [format string syntax](https://docs.python.org/3/library/string.html#formatstrings) here, e.g. `{chunk:07d}`.
 
+File format is determined by a file extension in the URL. For example, if `S3PIPELINE_URL` ends with `.json` or `.json.gz`, JSON format is used. See Scrapy's built-in [FEED_EXPORTERS](https://docs.scrapy.org/en/latest/topics/feed-exports.html#std-setting-FEED_EXPORTERS) settings for supported formats. If the file extension is not available in `FEED_EXPORTERS`, JSONLines format is used by default.
+
 ### S3PIPELINE_MAX_CHUNK_SIZE (Optional)
 
 Default: `100`
diff --git a/s3pipeline/pipelines.py b/s3pipeline/pipelines.py
index d9835d1..70e2256 100644
--- a/s3pipeline/pipelines.py
+++ b/s3pipeline/pipelines.py
@@ -3,8 +3,9 @@ from urllib.parse import urlparse
 from datetime import datetime
 import gzip
 from threading import Timer
+import os.path
 
-from scrapy.exporters import JsonLinesItemExporter
+from scrapy.utils.misc import load_object
 
 from s3pipeline.strategies.error import UploadError
 
@@ -29,6 +30,11 @@ class S3Pipeline:
         self.use_gzip = settings.getbool('S3PIPELINE_GZIP', url.endswith('.gz'))
         self.max_wait_upload_time = settings.getfloat('S3PIPELINE_MAX_WAIT_UPLOAD_TIME', 30.0)
 
+        uncompressed_url = url[:-3] if url.endswith('.gz') else url
+        _, ext = os.path.splitext(uncompressed_url)
+        feed_exporter_key = ext[1:].lower()
+        self.exporter_cls = load_object(settings.getwithbase('FEED_EXPORTERS').get(feed_exporter_key, 'scrapy.exporters.JsonLinesItemExporter'))
+
         if o.scheme == 's3':
             from .strategies.s3 import S3Strategy
             self.strategy = S3Strategy(settings)
@@ -120,7 +126,7 @@ class S3Pipeline:
         f = gzip.GzipFile(mode='wb', fileobj=bio) if self.use_gzip else bio
 
         # Build file object using ItemExporter
-        exporter = JsonLinesItemExporter(f, encoding='utf-8')
+        exporter = self.exporter_cls(f, encoding='utf-8')
         exporter.start_exporting()
         for item in self.items:
             exporter.export_item(item)


Test patch:
diff --git a/tests/test_pipelines.py b/tests/test_pipelines.py
index 3ad8c01..9e2d92b 100644
--- a/tests/test_pipelines.py
+++ b/tests/test_pipelines.py
@@ -1,36 +1,105 @@
 from unittest import TestCase
 
-from scrapy.settings import BaseSettings
+from scrapy.exporters import JsonLinesItemExporter, JsonItemExporter
+from scrapy.settings import BaseSettings, default_settings
 
 from s3pipeline import S3Pipeline
 from s3pipeline.strategies.s3 import S3Strategy
 from s3pipeline.strategies.gcs import GCSStrategy
 
-class TestPipelinesS3(TestCase):
-    def setUp(self):
-        self.settings = BaseSettings({
+class TestPipelineSettings(TestCase):
+
+    def test_s3(self):
+        settings = BaseSettings({
             'S3PIPELINE_URL': 's3://my-bucket/{name}/{time}/items.{chunk:07d}.jl.gz',
+            'FEED_EXPORTERS_BASE': default_settings.FEED_EXPORTERS_BASE,
         })
 
-    def test_settings(self):
-        pipeline = S3Pipeline(self.settings, None)
+        pipeline = S3Pipeline(settings, None)
         self.assertEqual(pipeline.bucket_name, 'my-bucket')
         self.assertEqual(pipeline.object_key_template, '{name}/{time}/items.{chunk:07d}.jl.gz')
         self.assertEqual(pipeline.max_chunk_size, 100)
         self.assertTrue(pipeline.use_gzip)
+        self.assertEqual(pipeline.max_wait_upload_time, 30)
         self.assertIsInstance(pipeline.strategy, S3Strategy)
+        self.assertEqual(pipeline.exporter_cls, JsonLinesItemExporter)
 
 
-class TestPipelinesGCS(TestCase):
-    def setUp(self):
-        self.settings = BaseSettings({
+    def test_gcs(self):
+        settings = BaseSettings({
             'S3PIPELINE_URL': 'gs://my-bucket/{name}/{time}/items.{chunk:07d}.jl',
+            'FEED_EXPORTERS_BASE': default_settings.FEED_EXPORTERS_BASE,
         })
 
-    def test_settings(self):
-        pipeline = S3Pipeline(self.settings, None)
+        pipeline = S3Pipeline(settings, None)
         self.assertEqual(pipeline.bucket_name, 'my-bucket')
         self.assertEqual(pipeline.object_key_template, '{name}/{time}/items.{chunk:07d}.jl')
         self.assertEqual(pipeline.max_chunk_size, 100)
         self.assertFalse(pipeline.use_gzip)
+        self.assertEqual(pipeline.max_wait_upload_time, 30)
         self.assertIsInstance(pipeline.strategy, GCSStrategy)
+        self.assertEqual(pipeline.exporter_cls, JsonLinesItemExporter)
+
+    def test_json(self):
+        settings = BaseSettings({
+            'S3PIPELINE_URL': 's3://my-bucket/{name}/{time}/items.{chunk:07d}.json',
+            'FEED_EXPORTERS_BASE': default_settings.FEED_EXPORTERS_BASE,
+        })
+
+        pipeline = S3Pipeline(settings, None)
+        self.assertFalse(pipeline.use_gzip)
+        self.assertEqual(pipeline.exporter_cls, JsonItemExporter)
+
+    def test_json_gz(self):
+        settings = BaseSettings({
+            'S3PIPELINE_URL': 's3://my-bucket/{name}/{time}/items.{chunk:07d}.json.gz',
+            'FEED_EXPORTERS_BASE': default_settings.FEED_EXPORTERS_BASE,
+        })
+
+        pipeline = S3Pipeline(settings, None)
+        self.assertTrue(pipeline.use_gzip)
+        self.assertEqual(pipeline.exporter_cls, JsonItemExporter)
+
+    def test_force_gzip(self):
+
+        settings = BaseSettings({
+            'S3PIPELINE_URL': 's3://my-bucket/{name}/{time}/items.{chunk:07d}.jl',
+            'S3PIPELINE_GZIP': True,
+            'FEED_EXPORTERS_BASE': default_settings.FEED_EXPORTERS_BASE,
+        })
+
+        pipeline = S3Pipeline(settings, None)
+        self.assertTrue(pipeline.use_gzip)
+
+    def test_force_no_gzip(self):
+
+        settings = BaseSettings({
+            'S3PIPELINE_URL': 's3://my-bucket/{name}/{time}/items.{chunk:07d}.jl.gz',
+            'S3PIPELINE_GZIP': False,
+            'FEED_EXPORTERS_BASE': default_settings.FEED_EXPORTERS_BASE,
+        })
+
+        pipeline = S3Pipeline(settings, None)
+        self.assertFalse(pipeline.use_gzip)
+
+    def test_max_chunk_size(self):
+
+        settings = BaseSettings({
+            'S3PIPELINE_URL': 's3://my-bucket/{name}/{time}/items.{chunk:07d}.jl.gz',
+            'S3PIPELINE_MAX_CHUNK_SIZE': 1000,
+            'FEED_EXPORTERS_BASE': default_settings.FEED_EXPORTERS_BASE,
+        })
+
+        pipeline = S3Pipeline(settings, None)
+        self.assertEqual(pipeline.max_chunk_size, 1000)
+
+    def test_max_wait_upload_time(self):
+
+        settings = BaseSettings({
+            'S3PIPELINE_URL': 's3://my-bucket/{name}/{time}/items.{chunk:07d}.jl.gz',
+            'S3PIPELINE_MAX_WAIT_UPLOAD_TIME': 300,
+            'FEED_EXPORTERS_BASE': default_settings.FEED_EXPORTERS_BASE,
+        })
+
+        pipeline = S3Pipeline(settings, None)
+        self.assertEqual(pipeline.max_wait_upload_time, 300)