Factoring move with requeue (basetasks/utils.py)

Author: DasFranck

mrq/basetasks/utils.py

@@ -98,30 +98,7 @@ def perform_action(self, action, query, destination_queue):
             if list(query.keys()) == ["queue"]:
                 Queue(query["queue"]).empty()
 
-        elif action == "move":
-            cursor = self.collection.find(query, projection=["_id", "queue"])
-            fetched_jobs = list(cursor)
-            for jobs in group_iter(fetched_jobs, n=1000):
-                jobs_by_queue = defaultdict(list)
-                for job in jobs:
-                    jobs_by_queue[job["queue"]].append(job["_id"])
-                    stats["requeued"] += 1
-
-                for queue in jobs_by_queue:
-
-                    updates = {
-                        "status": "queued",
-                        "datequeued": datetime.datetime.utcnow(),
-                        "dateupdated": datetime.datetime.utcnow(),
-                        "queue": destination_queue,
-                        "retry_count": 0
-                    }
-
-                    self.collection.update({
-                        "_id": {"$in": jobs_by_queue[queue]}
-                    }, {"$set": updates}, multi=True)
-
-        elif action in ("requeue", "requeue_retry"):
+        elif action in ("requeue", "requeue_retry", "move"):
 
             # Requeue task by groups of maximum 1k items (if all in the same
             # queue)
@@ -150,7 +127,7 @@ def perform_action(self, action, query, destination_queue):
                     if destination_queue is not None:
                         updates["queue"] = destination_queue
 
-                    if action == "requeue":
+                    if action in ("requeue", "move"):
                         updates["retry_count"] = 0
 
                     self.collection.update({