use more explicit name approach to mark 'target_queue' for specific celery queue tasks

Author: mkovalua

api/share/utils.py

@@ -50,24 +50,31 @@ def is_qa_resource(resource):
     return has_qa_tags or has_qa_title
 
 
-def update_share(resource, urgent=True):
+def update_share(resource, target_queue=None):
+    """
+    By default, tasks are routed to queue based on module routing in CeleryRouter,
+    :param resource: osf resource that is needed to be reindexed
+    :param target_queue: should be task queue attribute of CeleryConfig f.e 'task_low_queue' for bulk background
+    passing 'target_queue' allows low-level queue task run (reindexing files after a user merge) even though
+    related module path may be marked to work with task_high_queue.
+    """
     if not settings.SHARE_ENABLED:
         return
     if not hasattr(resource, 'guids'):
         logger.error(f'update_share called on non-guid resource: {resource}')
         return
-    if urgent:
-        _enqueue_update_share(resource)
+    if target_queue is not None:
+        _enqueue_update_share(resource, target_queue)
     else:
-        _enqueue_update_share(resource, urgent=False)
+        _enqueue_update_share(resource)
 
 
-def _enqueue_update_share(osfresource, urgent=True):
+def _enqueue_update_share(osfresource, target_queue=None):
     _osfguid_value = osfresource.guids.values_list('_id', flat=True).first()
     if not _osfguid_value:
         logger.warning(f'update_share skipping resource that has no guids: {osfresource}')
         return
-    enqueue_task(task__update_share.s(_osfguid_value, urgent=urgent))
+    enqueue_task(task__update_share.s(_osfguid_value, target_queue=target_queue))
 
 
 @celery_app.task(
@@ -76,7 +83,7 @@ def _enqueue_update_share(osfresource, urgent=True):
     max_retries=4,
     retry_backoff=True,
 )
-def task__update_share(self, guid: str, is_backfill=False, osfmap_partition_name='MAIN', urgent=True):
+def task__update_share(self, guid: str, is_backfill=False, osfmap_partition_name='MAIN', target_queue=None):
     """
     Send SHARE/trove current metadata record(s) for the osf-guid-identified object
     """

framework/celery_tasks/routers.py

@@ -32,8 +32,8 @@ def route_for_task(self, task, args=None, kwargs=None):
         :param str task:    Of the form 'full.module.path.to.class.function'
         :returns dict:      Tells celery into which queue to route this task.
         """
-        if kwargs and (kwargs.get('urgent') is False):
-            return {'queue': CeleryConfig.task_low_queue}
+        if kwargs and (target_queue := kwargs.get('target_queue')):
+            return {'queue': target_queue}
         return {
             'queue': match_by_module(task)
         }

osf/models/user.py

@@ -68,6 +68,7 @@
 from website import filters
 from website.project import new_bookmark_collection
 from website.util.metrics import OsfSourceTags, unregistered_created_source_tag
+from website.settings import CeleryConfig
 from importlib import import_module
 from osf.models.notification_type import NotificationTypeEnum
 from osf.utils.requests import string_type_request_headers
@@ -894,12 +895,12 @@ def merge_user(self, user):
         )
         for file in nodes_files_to_reindex.iterator(chunk_size=100):
             try:
-                update_share(file, urgent=False)
+                update_share(file, target_queue=CeleryConfig.task_low_queue)
             except Exception as e:
                 logger.exception(f'Failed to SHARE reindex file {file._id} during user merge: {e}')
         for file in preprints_files_to_reindex.iterator(chunk_size=100):
             try:
-                update_share(file, urgent=False)
+                update_share(file, target_queue=CeleryConfig.task_low_queue)
             except Exception as e:
                 logger.exception(f'Failed to SHARE reindex preprints file {file._id} during user merge: {e}')