schemas: trigger update on ES and reindexing when schema mapping changed

annatrz · annatrz · commit e6089aa2c4e5 · 2019-11-15T14:47:47.000+01:00
* signals on mappings update
* recreate indices on mappings update (del -&gt; create)
* reindex all records belonging to updated schema
* change ```cap fixtures schemas``` to update schemas in the db (before was
just skipping if schema already existed in the db), triggering ES
changes in reindexing

Signed-off-by: Anna Trzcinska &lt;anna.trzcinska@cern.ch&gt;
diff --git a/cap/modules/deposit/ext.py b/cap/modules/deposit/ext.py
@@ -1,13 +1,16 @@
 """Initialize extension."""
 
 from __future__ import absolute_import, print_function
-from cap.modules.schemas.models import Schema
+
 from invenio_search import current_search
 
+from cap.modules.schemas.models import Schema
+
+from .receivers import handle_deposit_mapping_updated
+
 
 class CAPDeposit(object):
     """CAPDeposit extension."""
-
     def __init__(self, app=None):
         """Extension initialization."""
         if app:
diff --git a/cap/modules/deposit/receivers.py b/cap/modules/deposit/receivers.py
@@ -0,0 +1,35 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of CERN Analysis Preservation Framework.
+# Copyright (C) 2016 CERN.
+#
+# CERN Analysis Preservation Framework is free software; you can redistribute
+# it and/or modify it under the terms of the GNU General Public License as
+# published by the Free Software Foundation; either version 2 of the
+# License, or (at your option) any later version.
+#
+# CERN Analysis Preservation Framework is distributed in the hope that it will
+# be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with CERN Analysis Preservation Framework; if not, write to the
+# Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston,
+# MA 02111-1307, USA.
+#
+# In applying this license, CERN does not
+# waive the privileges and immunities granted to it by virtue of its status
+# as an Intergovernmental Organization or submit itself to any jurisdiction.
+"""Registered signal handlers for records module."""
+from invenio_jsonschemas.proxies import current_jsonschemas
+
+from cap.modules.records.utils import reindex_by_schema_url
+from cap.modules.schemas.signals import deposit_mapping_updated
+
+
+@deposit_mapping_updated.connect
+def handle_deposit_mapping_updated(schema):
+    """Reindex all the deposits when mapping in ES got updated."""
+    schema_url = current_jsonschemas.path_to_url(schema.deposit_path)
+    reindex_by_schema_url(schema_url, 'depid')
diff --git a/cap/modules/records/__init__.py b/cap/modules/records/__init__.py
@@ -21,5 +21,6 @@
 # In applying this license, CERN does not
 # waive the privileges and immunities granted to it by virtue of its status
 # as an Intergovernmental Organization or submit itself to any jurisdiction.
-
 """Data model package."""
+
+from .receivers import handle_record_mapping_updated
diff --git a/cap/modules/records/receivers.py b/cap/modules/records/receivers.py
@@ -0,0 +1,37 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of CERN Analysis Preservation Framework.
+# Copyright (C) 2016 CERN.
+#
+# CERN Analysis Preservation Framework is free software; you can redistribute
+# it and/or modify it under the terms of the GNU General Public License as
+# published by the Free Software Foundation; either version 2 of the
+# License, or (at your option) any later version.
+#
+# CERN Analysis Preservation Framework is distributed in the hope that it will
+# be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with CERN Analysis Preservation Framework; if not, write to the
+# Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston,
+# MA 02111-1307, USA.
+#
+# In applying this license, CERN does not
+# waive the privileges and immunities granted to it by virtue of its status
+# as an Intergovernmental Organization or submit itself to any jurisdiction.
+"""Registered signal handlers for deposit module."""
+from invenio_jsonschemas.proxies import current_jsonschemas
+
+from cap.modules.schemas.signals import record_mapping_updated
+
+from .utils import reindex_by_schema_url
+
+
+@record_mapping_updated.connect
+def handle_record_mapping_updated(schema):
+    """Reindex all the record when mapping in ES got updated."""
+    schema_url = current_jsonschemas.path_to_url(schema.record_path)
+
+    reindex_by_schema_url(schema_url, 'recid')
diff --git a/cap/modules/records/utils.py b/cap/modules/records/utils.py
@@ -27,9 +27,15 @@
 import string
 
 from flask import url_for
+from invenio_db import db
+from invenio_indexer.api import RecordIndexer
 from invenio_pidstore.errors import PIDDoesNotExistError
-from invenio_pidstore.models import PersistentIdentifier
+from invenio_pidstore.models import PersistentIdentifier, PIDStatus
+from invenio_records.models import RecordMetadata
 from six.moves.urllib import parse
+from sqlalchemy import cast
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.dialects.sqlite import JSON
 
 
 def generate_recid(experiment):
@@ -71,3 +77,27 @@ def api_url_for(endpoint, pid, **kwargs):
                   **kwargs)
 
     return url_to_api_url(url)
+
+
+def reindex_by_schema_url(schema_url, pid_type):
+    """Reindex all records of given pid_type belonging to that schema."""
+    def _get_json_type():
+        """If postgres db return JSONB, else JSON."""
+        return JSONB if db.session.bind.dialect.name == 'postgresql' else JSON
+
+    indexer = RecordIndexer()
+
+    ids = (x[0] for x in RecordMetadata.query.filter(
+        RecordMetadata.json['$schema'] == cast(
+            schema_url, _get_json_type())).values(RecordMetadata.id))
+
+    filtered_by_pid_type = (x[0] for x in PersistentIdentifier.query.filter(
+        PersistentIdentifier.object_type == 'rec',
+        PersistentIdentifier.pid_type == pid_type, PersistentIdentifier.status
+        == PIDStatus.REGISTERED, PersistentIdentifier.object_uuid.in_(
+            ids)).values(PersistentIdentifier.object_uuid))
+
+    print('{} records will be reindexed...'.format(schema_url))
+
+    indexer.bulk_index(filtered_by_pid_type)
+    indexer.process_bulk_queue(es_bulk_kwargs={'raise_on_error': True})
diff --git a/cap/modules/schemas/cli.py b/cap/modules/schemas/cli.py
@@ -69,18 +69,18 @@ def add_schema_from_fixture(data=None):
         with db.session.begin_nested():
             with db.session.begin_nested():
                 try:
-                    schema = Schema.get(name=data['name'],
-                                        version=data['version'])
-                    click.secho('{} already exist in the db.'.format(
-                        str(name)))
-                    return
+                    schema = Schema.get(name=name, version=data['version'])
+                    schema.update(**data)
+                    msg, fg = '{} updated.'.format(str(name)), 'green'
 
                 except JSONSchemaNotFound:
-                    schema = Schema(**data)
-                    db.session.add(schema)
+                    db.session.add(Schema(**data))
+                    msg, fg = '{} added.'.format(str(name)), 'green'
 
             if allow_all:
                 schema.add_read_access_for_all_users()
+            else:
+                schema.revoke_access_for_all_users()
 
     except IntegrityError:
         click.secho('Error occured during adding {} to the db. \n'.format(
@@ -89,4 +89,4 @@ def add_schema_from_fixture(data=None):
         return
 
     db.session.commit()
-    click.secho('{} has been added.'.format(str(name)), fg='green')
+    click.secho(msg, fg=fg)
diff --git a/cap/modules/schemas/models.py b/cap/modules/schemas/models.py
@@ -44,6 +44,7 @@
 
 from .permissions import SchemaAdminAction, SchemaReadAction
 from .serializers import resolved_schemas_serializer, schema_serializer
+from .signals import deposit_mapping_updated, record_mapping_updated
 
 ES_FORBIDDEN = r' ,"\<*>|?'
 
@@ -208,11 +209,30 @@ def add_read_access_for_all_users(self):
         """Give read access to all authenticated users."""
         assert self.id
 
-        db.session.add(
-            ActionSystemRoles.allow(SchemaReadAction(self.id),
-                                    role=authenticated_user))
+        try:
+            ActionSystemRoles.query.filter(
+                ActionSystemRoles.action == 'schema-object-read',
+                ActionSystemRoles.argument == str(self.id),
+                ActionSystemRoles.role_name == 'authenticated_user').one()
+        except NoResultFound:
+            db.session.add(
+                ActionSystemRoles.allow(SchemaReadAction(self.id),
+                                        role=authenticated_user))
         db.session.flush()
 
+    def revoke_access_for_all_users(self):
+        """Revoke read access to all authenticated users."""
+        assert self.id
+
+        try:
+            db.session.delete(
+                ActionSystemRoles.query.filter(
+                    ActionSystemRoles.action == 'schema-object-read',
+                    ActionSystemRoles.argument == str(self.id),
+                    ActionSystemRoles.role_name == 'authenticated_user').one())
+        except NoResultFound:
+            pass
+
     def give_admin_access_for_user(self, user):
         """Give admin access for users."""
         assert self.id
@@ -270,25 +290,6 @@ def name_to_es_name(name):
     return name.replace('/', '-')
 
 
-def create_index(index_name, mapping_body, aliases):
-    """Create index in elasticsearch, add under given aliases."""
-    if not es.indices.exists(index_name):
-        current_search.mappings[index_name] = {}  # invenio search needs it
-
-        es.indices.create(index=index_name,
-                          body={'mappings': mapping_body},
-                          ignore=False)
-
-        for alias in aliases:
-            es.indices.update_aliases(
-                {'actions': [{
-                    'add': {
-                        'index': index_name,
-                        'alias': alias
-                    }
-                }]})
-
-
 @event.listens_for(Schema, 'after_insert')
 def after_insert_schema(target, value, schema):
     """On schema insert, create corresponding indexes and aliases in ES."""
@@ -303,6 +304,43 @@ def after_insert_schema(target, value, schema):
         current_cache.delete_memoized(import_string(mappings_imp))
 
 
+@event.listens_for(Schema.deposit_mapping, 'set')
+def after_deposit_mapping_updated(target, value, oldvalue, initiator):
+    """If deposit mapping field was updated:
+        * trigger mapping update in ES
+        * send signal
+        """
+    if target.is_indexed:
+        if es.indices.exists(target.deposit_index):
+            es.indices.delete(index=target.deposit_index)
+
+        create_index(target.deposit_index, value, target.deposit_aliases)
+        deposit_mapping_updated.send(target)
+
+        if target.use_deposit_as_record:
+            if es.indices.exists(target.record_index):
+                es.indices.delete(index=target.record_index)
+
+            create_index(target.record_index, value, target.record_aliases)
+
+            record_mapping_updated.send(target)
+
+
+@event.listens_for(Schema.record_mapping, 'set')
+def after_record_mapping_updated(target, value, oldvalue, initiator):
+    """If record mapping field was updated:
+        * trigger mapping update in ES
+        * send signal
+        """
+    if target.is_indexed and not target.use_deposit_as_record:
+        if es.indices.exists(target.record_index):
+            es.indices.delete(index=target.record_index)
+
+        create_index(target.record_index, value, target.record_aliases)
+
+        record_mapping_updated.send(target)
+
+
 @event.listens_for(Schema, 'after_delete')
 def before_delete_schema(mapper, connect, schema):
     """On schema delete, delete corresponding indexes and aliases in ES."""
@@ -316,7 +354,20 @@ def before_delete_schema(mapper, connect, schema):
             current_cache.delete_memoized(import_string(mappings_imp))
 
 
-@db.event.listens_for(Schema, 'before_update', propagate=True)
-def timestamp_before_update(mapper, connection, target):
-    """Update `updated` property with current time on `before_update` event."""
-    target.updated = datetime.utcnow()
+def create_index(index_name, mapping_body, aliases):
+    """Create index in elasticsearch, add under given aliases."""
+    if not es.indices.exists(index_name):
+        current_search.mappings[index_name] = {}  # invenio search needs it
+
+        es.indices.create(index=index_name,
+                          body={'mappings': mapping_body},
+                          ignore=False)
+
+        for alias in aliases:
+            es.indices.update_aliases(
+                {'actions': [{
+                    'add': {
+                        'index': index_name,
+                        'alias': alias
+                    }
+                }]})
diff --git a/cap/modules/schemas/signals.py b/cap/modules/schemas/signals.py
@@ -0,0 +1,34 @@
+# -*- coding: utf-8 -*-
+#
+# This file is part of CERN Analysis Preservation Framework.
+# Copyright (C) 2016, 2017 CERN.
+#
+# CERN Analysis Preservation Framework is free software; you can redistribute
+# it and/or modify it under the terms of the GNU General Public License as
+# published by the Free Software Foundation; either version 2 of the
+# License, or (at your option) any later version.
+#
+# CERN Analysis Preservation Framework is distributed in the hope that it will
+# be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with CERN Analysis Preservation Framework; if not, write to the
+# Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston,
+# MA 02111-1307, USA.
+#
+# In applying this license, CERN does not
+# waive the privileges and immunities granted to it by virtue of its status
+# as an Intergovernmental Organization or submit itself to any jurisdiction.
+"""Signals for schemas module."""
+
+from __future__ import absolute_import, print_function
+
+from blinker import Namespace
+
+_signals = Namespace()
+
+deposit_mapping_updated = _signals.signal('deposit_mapping-updated')
+
+record_mapping_updated = _signals.signal('record_mapping-updated')
diff --git a/setup.py b/setup.py
@@ -76,7 +76,7 @@
     # temporary pinned since there are 'connection closed' issues
     # on production server
     'urllib3[secure]==1.22',
-    'SQLAlchemy-Continuum==1.3.4',
+    'sqlalchemy==1.3',
 
     # temporary pinned since there are 'fs' conslicts between
     # 'reana-commons' and 'invenio-files-rest'
@@ -95,15 +95,13 @@
     # "raven" versions needed till we FIX dependecies on installation
     'raven[flask]>=5.0.0,<5.5',
     'invenio-logging[sentry]>=1.0.0b1',
-
     'uWSGI==2.0.17',
     'uwsgi-tools==1.1.1',
     'uwsgitop==0.10',
 ]
 
 packages = find_packages()
 
-
 # Get the version string. Cannot be done with import!
 g = {}
 with open(os.path.join('cap', 'version.py'), 'rt') as fp:
diff --git a/tests/integration/test_schemas_views.py b/tests/integration/test_schemas_views.py
diff --git a/tests/unit/test_views.py b/tests/unit/test_views.py
