Add ability to specify queue for indexing
@@ -32,17 +32,25 @@ class Command(BaseCommand):
}
yield index_objects_to_es.si(**data)
- def _run_reindex_tasks(self, models):
+ def _run_reindex_tasks(self, models, queue):
+ apply_async_kwargs = {'priority': 0}
+ if queue:
+ log.info('Adding indexing tasks to queue {0}'.format(queue))
+ apply_async_kwargs['queue'] = queue
+ else:
+ log.info('Adding indexing tasks to default queue')
+
+ index_time = timezone.now()
+ timestamp = datetime.datetime.now().strftime('%Y%m%d%H%M%S')
+
for doc in registry.get_documents(models):
queryset = doc().get_queryset()
# Get latest object from the queryset
- index_time = timezone.now()
app_label = queryset.model._meta.app_label
model_name = queryset.model.__name__
index_name = doc._doc_type.index
- timestamp = datetime.datetime.now().strftime('%Y%m%d%H%M%S')
new_index_name = "{}_{}".format(index_name, timestamp)
pre_index_task = create_new_es_index.si(app_label=app_label,
@@ -69,14 +77,22 @@ class Command(BaseCommand):
# http://celery.readthedocs.io/en/latest/userguide/canvas.html#chords
chord_tasks = chord(header=indexing_tasks, body=post_index_task)
+ if queue:
+ chord_tasks.set(queue=queue)
# http://celery.readthedocs.io/en/latest/userguide/canvas.html#chain
- chain(pre_index_task, chord_tasks, missed_index_task).apply_async()
+ chain(pre_index_task, chord_tasks, missed_index_task).apply_async(**apply_async_kwargs)
message = ("Successfully issued tasks for {}.{}, total {} items"
.format(app_label, model_name, queryset.count()))
log.info(message)
def add_arguments(self, parser):
+ parser.add_argument(
+ '--queue',
+ dest='queue',
+ action='store',
+ help="Set the celery queue name for the task."
+ )
parser.add_argument(
'--models',
dest='models',
@@ -98,4 +114,8 @@ class Command(BaseCommand):
if options['models']:
models = [apps.get_model(model_name) for model_name in options['models']]
- self._run_reindex_tasks(models=models)
+ queue = None
+ if options.get('queue'):
+ queue = options['queue']
+
+ self._run_reindex_tasks(models=models, queue=queue)
