[#9] Index datapackage in elasticsearch at theend of the pipeline

fixes #9
datopian · Aug 4, 2017 · 8d0f039 · 8d0f039
1 parent f1edbe8
commit 8d0f039
Show file tree

Hide file tree

Showing 4 changed files with 66 additions and 2 deletions.
diff --git a/datapackage_pipelines_assembler/generator.py b/datapackage_pipelines_assembler/generator.py
@@ -54,7 +54,8 @@ def generate_pipeline(cls, source):
                         'stats': {
                             'rowcount': 0,
                             'bytes': 0,
-                        }
+                        },
+                        'id': pipeline_id
                     }
                 },
                 {
@@ -113,6 +114,22 @@ def generate_pipeline(cls, source):
                             "resource-hash": "hash",
                         }
                     }
-                }
+                },
+                {
+                    'run': 'assembler.add_indexing_resource'
+                },
+                {
+                    'run': 'elasticsearch.dump.to_index',
+                    'parameters': {
+                        'indexes': {
+                            'datasets': [
+                                {
+                                    'resource-name': 'datasets',
+                                    'doc-type': 'dataset'
+                                }
+                            ]
+                        }
+                    }
+                },
             ]
         }
diff --git a/datapackage_pipelines_assembler/processors/add_indexing_resource.py b/datapackage_pipelines_assembler/processors/add_indexing_resource.py
@@ -0,0 +1,44 @@
+import os
+
+import copy
+import datapackage
+import itertools
+
+from datapackage_pipelines.wrapper import ingest, spew
+
+parameters, dp, res_iter = ingest()
+
+
+def modify_datapackage(dp):
+    dp['resources'].append({
+        'name': 'datasets',
+        'path': 'nonexistent',
+        'schema': {
+            'fields': [
+                {'name': 'id', 'type': 'string'},
+                {'name': 'name', 'type': 'string'},
+                {'name': 'title', 'type': 'string'},
+                {'name': 'description', 'type': 'string'},
+                {'name': 'resources', 'type': 'array',
+                 'es:itemType': 'object', 'es:index': False},
+                {'name': 'datahub', 'type': 'object',
+                 'es:schema': {
+                    'fields': [
+                        {'name': 'owner', 'type': 'string'},
+                        {'name': 'ownerid', 'type': 'string'},
+                    ]
+                 }
+                },
+            ],
+            'primaryKey': ['id']
+        }
+    })
+    return dp
+
+
+def dataset_resource(dp):
+    yield dp
+
+
+spew(modify_datapackage(dp),
+     itertools.chain(res_iter, [dataset_resource(dp)]))
diff --git a/datapackage_pipelines_assembler/processors/update_metadata.py b/datapackage_pipelines_assembler/processors/update_metadata.py
@@ -2,6 +2,8 @@
 
 
 def modify_datapackage(dp, parameters, stats):
+    id = parameters.pop('id')
+    dp['id'] = id
     dp['datahub'] = parameters
     return dp
 

diff --git a/setup.py b/setup.py
@@ -19,6 +19,7 @@ def read(*paths):
 NAME = PACKAGE.replace('_', '-')
 INSTALL_REQUIRES = [
     'datapackage-pipelines',
+    'datapackage-pipelines-elasticsearch',
     'psycopg2',
     'tweepy',
     'facebook-sdk',