Merge pull request #32 from HBS-HBX/#31_es_update_newer_workers_info

fixes #31 make `es_update --newer --workers` receive workers param

Author: codekiln

CHANGELOG.rst

@@ -1,6 +1,10 @@
 Changelog
 ---------
 
+0.7.3 (2018-08-14)
+~~~~~~~~~~~~~~~~~~
+* fix #31 es_update movies --newer --workers does not store worker information
+
 0.7.2 (2018-08-13)
 ~~~~~~~~~~~~~~~~~~
 * fix #21 wrong batch update total using multiprocessing in 0.7.1

django_elastic_migrations/__init__.py

@@ -10,7 +10,7 @@
 from django_elastic_migrations.utils import loading
 from django_elastic_migrations.utils.django_elastic_migrations_log import get_logger
 
-__version__ = '0.7.2'
+__version__ = '0.7.3'
 
 default_app_config = 'django_elastic_migrations.apps.DjangoElasticMigrationsConfig'  # pylint: disable=invalid-name
 

django_elastic_migrations/models.py

@@ -962,10 +962,11 @@ def __init__(self, *args, **kwargs):
         if NewerModeMixin.MODE_NAME in kwargs:
             self.self_kwargs[NewerModeMixin.MODE_NAME] = True
 
-        # retain a history of how this command was called
-        self.task_kwargs = json.dumps(self.self_kwargs, sort_keys=True)
-
         super(UpdateIndexAction, self).__init__(*args, **kwargs)
+        if self.task_kwargs == '{}' and self.self_kwargs:
+            # retain a history of how this command was called
+            self.task_kwargs = json.dumps(self.self_kwargs, sort_keys=True)
+
         self._batch_num = 0
         self._expected_remaining = 0
         self._indexed_docs = 0
@@ -1103,6 +1104,14 @@ def perform_action(self, dem_index, *args, **kwargs):
             use_self_dict_format=True
         )
 
+    def apply_to_newer(self, given_version, action=None):
+        versions = self.index.get_newer_versions(given_version=given_version)
+        kwargs = deepcopy(self.self_kwargs)
+        # we don't want child update index actions to also do 'newer' tasks
+        kwargs.pop(NewerModeMixin.MODE_NAME)
+        update_index_action = UpdateIndexAction(**kwargs)
+        super(UpdateIndexAction, self).apply_to_newer(versions, update_index_action)
+
     def prepare_action(self, dem_index):
         self._index_name = self.index.name
         self._index_version_id = dem_index.get_version_id()
@@ -1119,12 +1128,8 @@ def prepare_action(self, dem_index):
                 raise IllegalDEMIndexState(msg)
 
             if self.newer_mode:
-                versions = self.index.get_newer_versions(given_version=index_version)
-                kwargs = deepcopy(self.self_kwargs)
-                # we don't want child update index actions to also do 'newer' tasks
-                kwargs.pop(NewerModeMixin.MODE_NAME)
-                update_index_action = UpdateIndexAction(**kwargs)
-                self.apply_to_newer(versions, update_index_action)
+                self.apply_to_newer(index_version)
+                return
             else:
                 self.index_version = index_version
                 self._index_version_name = index_version.name
@@ -1157,9 +1162,7 @@ def prepare_action(self, dem_index):
                 raise NoActiveIndexVersion(msg)
 
             if self.newer_mode:
-                versions = self.index.get_newer_versions(given_version=active_version)
-                self.apply_to_newer(versions, UpdateIndexAction())
-                # we're done, because the newer versions will get their own actions
+                self.apply_to_newer(active_version)
                 return
 
             # we have an active version for this index. now do the update.