Merge d095848 into 9749c2f

invenio_indexer/api.py

@@ -24,7 +24,7 @@
 from sqlalchemy.orm.exc import NoResultFound
 
 from .proxies import current_record_to_index
-from .signals import before_record_index
+from .signals import before_record_index, before_record_index_arguments
 
 
 class Producer(KombuProducer):
@@ -103,7 +103,7 @@ def mq_routing_key(self):
     #
     # High-level API
     #
-    def index(self, record):
+    def index(self, record, **kwargs):
         """Index a record.
 
         The caller is responsible for ensuring that the record has already been
@@ -116,13 +116,18 @@ def index(self, record):
         """
         index, doc_type = self.record_to_index(record)
 
+        body = self._prepare_record(record, index, doc_type)
+
+        arguments = self._prepare_arguments(record, index, doc_type, **kwargs)
+
         return self.client.index(
             id=str(record.id),
             version=record.revision_id,
             version_type=self._version_type,
             index=index,
             doc_type=doc_type,
-            body=self._prepare_record(record, index, doc_type),
+            body=body,
+            **arguments
         )
 
     def index_by_id(self, record_uuid):
@@ -312,6 +317,41 @@ def _prepare_record(record, index, doc_type):
 
         return data
 
+    @staticmethod
+    def _prepare_arguments(record, index, doc_type, **kwargs):
+        """Prepare arguments for indexing.
+
+        :param record: The record to prepare.
+        :param index: The Elasticsearch index.
+        :param doc_type: The Elasticsearch document type.
+        :param **kwargs: Extra parameters.
+        :returns: The arguments for Elasticsearch.
+        """
+        if current_app.config['INDEXER_REPLACE_REFS']:
+            data = copy.deepcopy(record.replace_refs())
+        else:
+            data = record.dumps()
+
+        data['_created'] = pytz.utc.localize(record.created).isoformat() \
+            if record.created else None
+        data['_updated'] = pytz.utc.localize(record.updated).isoformat() \
+            if record.updated else None
+
+        arguments = {}
+
+        # Allow modification of arguments prior to sending to Elasticsearch.
+        before_record_index_arguments.send(
+            current_app._get_current_object(),
+            json=data,
+            record=record,
+            index=index,
+            doc_type=doc_type,
+            arguments=arguments,
+            **kwargs
+        )
+
+        return arguments
+
 
 class BulkRecordIndexer(RecordIndexer):
     r"""Provide an interface for indexing records in Elasticsearch.

invenio_indexer/signals.py

@@ -25,3 +25,16 @@
 - ``index``: The index in which the record will be indexed.
 - ``doc_type``: The doc_type for the record.
 """
+
+before_record_index_arguments = _signals.signal('before-record-index-args')
+"""Signal sent before a record is indexed.
+
+The sender is the current Flask application, and two keyword arguments are
+provided:
+
+- ``json``: The dumped record dictionary which can be modified.
+- ``record``: The record being indexed.
+- ``index``: The index in which the record will be indexed.
+- ``doc_type``: The doc_type for the record.
+- ``**kwargs``: Extra arguments.
+"""

setup.py

@@ -65,7 +65,6 @@
     'invenio-pidstore>=1.0.0',
     'invenio-records>=1.0.0',
     'pytz>=2016.4',
-    'redis>=2.10.0,<3.0.0',
 ]
 
 packages = find_packages()