indexer: add parameter extraction before indexing

cern_search_rest_api/config.py

@@ -78,6 +78,7 @@ INDEXER_DEFAULT_INDEX = os.getenv('CERN_SEARCH_DEFAULT_INDEX', 'cernsearch-test-
 
 SEARCH_MAPPINGS = [os.getenv('CERN_SEARCH_INSTANCE', 'cernsearch-test')]
 SEARCH_USE_EGROUPS = ast.literal_eval(os.getenv('CERN_SEARCH_USE_EGROUPS', 'True'))
+SEARCH_DOC_PIPELINES = ast.literal_eval(os.getenv('CERN_SEARCH_DOC_PIPELINES', '{}'))
 
 # Records REST configuration
 # ===========================

cern_search_rest_api/modules/cernsearch/ext.py

@@ -7,7 +7,10 @@
 # granted to it by virtue of its status as Intergovernmental Organization
 # or submit itself to any jurisdiction.
 
-from cern_search_rest_api.modules.cernsearch.views import build_blueprint, build_health_blueprint
+from invenio_indexer.signals import before_record_index_arguments
+
+from cern_search_rest_api.modules.cernsearch.indexer import csas_indexer_receiver
+from cern_search_rest_api.modules.cernsearch.views import build_blueprint
 
 
 class CERNSearch(object):
@@ -23,6 +26,7 @@ class CERNSearch(object):
         self.init_config(app)
         blueprint = build_blueprint(app)
         app.register_blueprint(blueprint)
+        before_record_index_arguments.connect(csas_indexer_receiver, sender=app)
         app.extensions["cern-search"] = self
 
     def init_config(self, app):

cern_search_rest_api/modules/cernsearch/indexer.py

+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+# Copyright (C) 2018, CERN
+# This software is distributed under the terms of the GNU General Public
+# Licence version 3 (GPL Version 3), copied verbatim in the file "COPYING".
+# In applying this license, CERN does not waive the privileges and immunities
+# granted to it by virtue of its status as Intergovernmental Organization
+# or submit itself to any jurisdiction.
+
+from flask import current_app
+
+
+def csas_indexer_receiver(sender, json=None, record=None, index=None,
+                     doc_type=None, arguments=None, **kwargs):
+
+    pipeline_mapping = current_app.config['SEARCH_DOC_PIPELINES']
+
+    if pipeline_mapping:
+        pipeline = pipeline_mapping.get(doc_type, None)
+
+        if pipeline:
+            arguments['pipeline'] = pipeline