Combine adding indexing resource with custom s3 dumper

fixes #8 fixes #9
datopian · Aug 5, 2017 · 260ed86 · 260ed86
1 parent fed5feb
commit 260ed86
Show file tree

Hide file tree

Showing 3 changed files with 20 additions and 8 deletions.
diff --git a/datapackage_pipelines_assembler/generator.py b/datapackage_pipelines_assembler/generator.py
@@ -4,7 +4,7 @@
 from datapackage_pipelines.generators import (
     GeneratorBase,
 )
-from .processors.add_indexing_resource import create_index
+from .processors.dump_to_s3 import create_index
 
 import logging
 log = logging.getLogger(__name__)
@@ -106,7 +106,7 @@ def generate_pipeline(cls, source):
                 #     }
                 # },
                 {
-                    'run': 'aws.dump.to_s3',
+                    'run': 'assembler.dump_to_s3',
                     'parameters': {
                         'force-format': False,
                         'handle-non-tabular': True,
@@ -122,9 +122,6 @@ def generate_pipeline(cls, source):
                         }
                     }
                 },
-                {
-                    'run': 'assembler.add_indexing_resource'
-                },
                 {
                     'run': 'elasticsearch.dump.to_index',
                     'parameters': {

diff --git a/...mbler/processors/add_indexing_resource.py → ...elines_assembler/processors/dump_to_s3.py b/...mbler/processors/add_indexing_resource.py → ...elines_assembler/processors/dump_to_s3.py
@@ -1,8 +1,11 @@
 import itertools
 
+import copy
 from tableschema_elasticsearch import Storage
 
 from datapackage_pipelines.wrapper import ingest, spew
+from datapackage_pipelines_aws.processors.dump.to_s3 import S3Dumper
+
 
 SCHEMA = {
     'fields': [
@@ -50,11 +53,22 @@ def dataset_resource(dp):
             'datahub'
         ]
     )
+    dp = copy.deepcopy(dp)
+    dp['resources'].pop()
     ret['datapackage'] = dp
     yield ret
 
 
+class MyS3Dumper(S3Dumper):
+
+    def prepare_datapackage(self, datapackage, params):
+        datapackage = super(MyS3Dumper, self).prepare_datapackage(datapackage, params)
+        return modify_datapackage(datapackage)
+
+    def handle_resources(self, datapackage, resource_iterator, parameters, stats):
+        yield from super(MyS3Dumper, self).handle_resources(datapackage, resource_iterator, parameters, stats)
+        yield [dataset_resource(datapackage)]
+
+
 if __name__ == "__main__":
-    parameters, dp, res_iter = ingest()
-    spew(modify_datapackage(dp),
-         itertools.chain(res_iter, [dataset_resource(dp)]))
+    MyS3Dumper()()
diff --git a/setup.py b/setup.py
@@ -20,6 +20,7 @@ def read(*paths):
 INSTALL_REQUIRES = [
     'datapackage-pipelines',
     'datapackage-pipelines-elasticsearch>=0.0.3',
+    'datapackage-pipelines-aws>=0.0.6',
     'psycopg2',
     'tweepy',
     'facebook-sdk',