Adding job moving feature

Author: DasFranck

mrq/basetasks/utils.py

@@ -98,6 +98,29 @@ 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"):
 
             # Requeue task by groups of maximum 1k items (if all in the same

mrq/dashboard/static/js/views/jobs.js

@@ -223,6 +223,15 @@ define(["jquery", "underscore", "views/generic/datatablepage", "models"],functio
         });
         self.refreshCallStack(job_id);
 
+      } else if (action == "move") {
+
+        var queue = prompt("Test");
+        self.jobaction(evt, {
+          "id": job_id,
+          "action": action,
+          "destination_queue": queue
+        });
+
       } else {
 
         self.jobaction(evt, {
@@ -392,6 +401,7 @@ define(["jquery", "underscore", "views/generic/datatablepage", "models"],functio
                   "<br/><br/>"+
                   "<button class='btn btn-xs btn-danger pull-right' data-action='cancel'><span class='glyphicon glyphicon-remove-circle'></span> Cancel</button>"+
                   "<button class='btn btn-xs btn-warning' data-action='requeue'><span class='glyphicon glyphicon-refresh'></span> Requeue</button>"+
+                  "<button class='btn btn-xs btn-warning' data-action='move'><span class='glyphicon glyphicon-refresh'></span> Move to...</button>"+
                 "</div>";
               }
               return "";

mrq/exceptions.py

@@ -27,6 +27,10 @@ class AbortInterrupt(_MrqInterrupt):
     pass
 
 
+class MovedInterrupt(_MrqInterrupt):
+    pass
+
+
 class RetryInterrupt(_MrqInterrupt):
     delay = None
     queue = None

mrq/job.py

@@ -5,7 +5,7 @@
 from bson import ObjectId
 from redis.exceptions import LockError
 import time
-from .exceptions import RetryInterrupt, MaxRetriesInterrupt, AbortInterrupt, MaxConcurrencyInterrupt
+from .exceptions import RetryInterrupt, MaxRetriesInterrupt, AbortInterrupt, MaxConcurrencyInterrupt, MovedInterrupt
 from .utils import load_class_by_path, group_iter
 import gevent
 import objgraph
@@ -270,6 +270,13 @@ def requeue(self, queue=None, retry_count=0):
             "retry_count": retry_count
         })
 
+    def move(self, queue=None, retry_count=0):
+        """ Cancel and requeues the current job in an other queue."""
+        exc = MovedInterrupt()
+        self._attach_original_exception(exc)
+        self.requeue(queue, retry_count)
+        raise exc
+
     def perform(self):
         """ Loads and starts the main task for this job, the saves the result. """