wip

invenio_users_resources/records/hooks.py

@@ -51,14 +51,20 @@ def post_commit(sender, session):
     # properties!
     sid = id(session)
 
-    reindex_user_roles = current_db_change_history.updated_users[sid]
-    reindex_group_roles = current_db_change_history.updated_roles[sid]
-    unindex_user_roles = current_db_change_history.deleted_users[sid]
-    unindex_group_roles = current_db_change_history.deleted_roles[sid]
+    if current_db_change_history.sessions.get(sid):
+        reindex_user_roles = current_db_change_history.sessions[sid].updated_users
+        reindex_group_roles = current_db_change_history.sessions[sid].updated_roles
 
-    reindex_user.delay(reindex_user_roles)
-    reindex_group.delay(reindex_group_roles)
-    unindex_user.delay(unindex_user_roles)
-    unindex_group.delay(unindex_group_roles)
+        unindex_user_roles = current_db_change_history.sessions[sid].deleted_users
+        unindex_group_roles = current_db_change_history.sessions[sid].deleted_roles
 
-    current_db_change_history._clear_dirty_sets(session)
+        # ----------------------------------------
+
+        reindex_user.delay(reindex_user_roles)
+        reindex_group.delay(reindex_group_roles)
+
+        unindex_user.delay(unindex_user_roles)
+        unindex_group.delay(unindex_group_roles)
+
+        current_db_change_history.sessions[sid].indexed = True
+        current_db_change_history.clear_session(sid)

invenio_users_resources/services/groups/tasks.py

@@ -20,24 +20,22 @@
 @shared_task(ignore_result=True)
 def reindex_group(role_ids):
     """Reindex the given user."""
-    for role_id in role_ids:
-        index = current_groups_service.record_cls.index
-        if current_groups_service.indexer.exists(index):
-            try:
-                group_agg = GroupAggregate.get_record(role_id)
-                current_groups_service.indexer.index(group_agg)
-            except search.exceptions.ConflictError as e:
-                current_app.logger.warn(f"Could not reindex group {role_id}: {e}")
+    index = current_groups_service.record_cls.index
+    if current_groups_service.indexer.exists(index):
+        try:
+            group_agg = [GroupAggregate.get_record(role_id) for role_id in role_ids]
+            current_groups_service.indexer.bulk_index(group_agg)
+        except search.exceptions.ConflictError as e:
+            current_app.logger.warn(f"Could not reindex group: {e}")
 
 
 @shared_task(ignore_result=True)
 def unindex_group(role_ids):
     """Unindex the given role/group."""
-    for role_id in role_ids:
-        index = current_groups_service.record_cls.index
-        if current_groups_service.indexer.exists(index):
-            try:
-                group_agg = GroupAggregate.get_record(role_id)
-                current_groups_service.indexer.delete(group_agg)
-            except search.exceptions.ConflictError as e:
-                current_app.logger.warn(f"Could not unindex group {role_id}: {e}")
+    index = current_groups_service.record_cls.index
+    if current_groups_service.indexer.exists(index):
+        try:
+            group_agg = [GroupAggregate.get_record(role_id) for role_id in role_ids]
+            current_groups_service.indexer.bulk_delete(group_agg)
+        except search.exceptions.ConflictError as e:
+            current_app.logger.warn(f"Could not unindex group: {e}")

invenio_users_resources/services/users/tasks.py

@@ -21,28 +21,32 @@
 @shared_task(ignore_result=True)
 def reindex_user(user_ids):
     """Reindex the given user."""
-    for user_id in user_ids:
-        index = current_users_service.record_cls.index
-        if current_users_service.indexer.exists(index):
-            try:
-                user_agg = UserAggregate.get_record(user_id)
-                current_users_service.indexer.index(user_agg)
-                # trigger reindexing of related records
+    index = current_users_service.record_cls.index
+    if current_users_service.indexer.exists(index):
+        try:
+            user_agg = [UserAggregate.get_record(user_id) for user_id in user_ids]
+            print("#" * 100)
+            print(user_agg)
+            current_users_service.indexer.bulk_index(user_agg)
+
+            # trigger reindexing of related records
+            for user in user_agg:
+                print(user)
                 send_change_notifications(
-                    "users", [(user_agg.id, str(user_agg.id), user_agg.revision_id)]
+                    "users", [(user.id, str(user.id), user.revision_id)]
                 )
-            except search.exceptions.ConflictError as e:
-                current_app.logger.warn(f"Could not reindex user {user_id}: {e}")
+            print("#" * 100)
+        except search.exceptions.ConflictError as e:
+            current_app.logger.warn(f"Could not reindex user: {e}")
 
 
 @shared_task(ignore_result=True)
 def unindex_user(user_ids):
     """Delete the given user from the index."""
-    for user_id in user_ids:
-        index = current_users_service.record_cls.index
-        if current_users_service.indexer.exists(index):
-            try:
-                user_agg = UserAggregate.get_record(user_id)
-                current_users_service.indexer.delete(user_agg)
-            except search.exceptions.ConflictError as e:
-                current_app.logger.warn(f"Could not unindex user {user_id}: {e}")
+    index = current_users_service.record_cls.index
+    if current_users_service.indexer.exists(index):
+        try:
+            user_agg = [UserAggregate.get_record(user_id) for user_id in user_ids]
+            current_users_service.indexer.bulk_delete(user_agg)
+        except search.exceptions.ConflictError as e:
+            current_app.logger.warn(f"Could not unindex user: {e}")