alembic: fix 05555c03fe49

* Uses Holding reindex to correct holdings items_count and public_items_count.

Co-Authored-by: Peter Weber <[email protected]>

rero_ils/alembic/05555c03fe49_correct_holdings_items_count.py

@@ -19,11 +19,7 @@
 """Correct holdings items_count and public_items_count."""
 from logging import getLogger
 
-from elasticsearch_dsl import Document
-from invenio_search import current_search_client
-
-from rero_ils.modules.holdings.api import HoldingsSearch
-from rero_ils.modules.items.api import ItemsSearch
+from rero_ils.modules.holdings.api import Holding, HoldingsSearch
 
 # revision identifiers, used by Alembic.
 revision = '05555c03fe49'
@@ -36,12 +32,14 @@
 
 def upgrade():
     """Upgrade index holdings."""
-    upgrade_downgrade('upgrade')
+    errors = upgrade_downgrade('upgrade')
+    LOGGER.info(f'upgraded to version: {revision} errors: {errors}')
 
 
 def downgrade():
     """Downgrade index holdings."""
-    upgrade_downgrade('downgrade')
+    errors = upgrade_downgrade('downgrade')
+    LOGGER.info(f'downgraded to version: {down_revision} errors: {errors}')
 
 
 def upgrade_downgrade(action):
@@ -50,47 +48,20 @@ def upgrade_downgrade(action):
     Correct items_count and public_items_count for holdings of type serial.
     :param str action: upgrade or downgrade.
     """
-    index = HoldingsSearch.Meta.index
     query = HoldingsSearch()\
         .filter('term', holdings_type='serial') \
         .source(['pid'])
 
     ids = [(h.meta.id, h.pid) for h in query.scan()]
-    count = 0
 
     LOGGER.info(f'Indexing {len(ids)} records ....')
-    for (_id, pid) in ids:
-        document = Document.get(_id, index=index, using=current_search_client)
-        items_count, public_items_count = get_counts(pid, action)
-
-        document.update(items_count=items_count,
-                        public_items_count=public_items_count,
-                        index=index,
-                        using=current_search_client,
-                        refresh=True)
-        count += 1
-        LOGGER.info(f'{count} records indexed.')
-
-
-def get_counts(pid, action):
-    """Calculate items_count and public_items_count.
-
-    :param str pid: holding pid.
-    :param str action: upgrade or downgrade.
-    :return: items_count and public_items_count
-    """
-    if action == 'upgrade':
-        item_search = ItemsSearch()[0:0]\
-                .filter('term', holding__pid=pid)\
-                .filter('term', issue__status="received")
-    else:
-        item_search = ItemsSearch()[0:0]\
-                .filter('term', holding__pid=pid)
-
-    items_count = item_search.count()
-    results = item_search.source([]).scan()
-    public_items_count = len([res for res in results
-                              if "_masked" not in res
-                              or not res['_masked']])
-
-    return items_count, public_items_count
+    errors = 0
+    for idx, (id, pid) in enumerate(ids):
+        LOGGER.info(f'{idx} * Reindex holding: {pid}.')
+        try:
+            hold = Holding.get_record_by_id(id)
+            hold.reindex()
+        except Exception as err:
+            LOGGER.error(f'{idx} * Reindex holding: {pid} {err}')
+            errors += 1
+    return errors

rero_ils/alembic/ce4923ba5286_delete_expired_items_temporary_locations.py

@@ -34,19 +34,25 @@ def index_items_with_temporary_location():
     query = ItemsSearch() \
         .filter('exists', field='temporary_location').source(['pid'])
     ids = [(hit.meta.id, hit.pid) for hit in query.scan()]
-    for id, pid in ids:
-        item = Item.get_record_by_id(id)
-        item.reindex()
-        LOGGER.info(f'  * Reindexed item#{pid}')
+    errors = 0
+    for idx, (id, pid) in enumerate(ids):
+        LOGGER.info(f'{idx} * Reindex item: {pid}')
+        try:
+            item = Item.get_record_by_id(id)
+            item.reindex()
+        except Exception as err:
+            LOGGER.error(f'{idx} * Reindex item: {pid} {err}')
+            errors += 1
+    return errors
 
 
 def upgrade():
     """Index items with temporary location."""
-    index_items_with_temporary_location()
-    LOGGER.info(f'upgraded to version: {revision}')
+    errors = index_items_with_temporary_location()
+    LOGGER.info(f'upgraded to version: {revision} errors: {errors}')
 
 
 def downgrade():
     """Index items with temporary location."""
-    index_items_with_temporary_location()
-    LOGGER.info(f'downgraded to version: {down_revision}')
+    errors = index_items_with_temporary_location()
+    LOGGER.info(f'downgraded to version: {down_revision} errors: {errors}')