Updates Changelog

Changelog

@@ -10,6 +10,8 @@
 3.0.1
 =====
 
+- Now depends on kombu 2.2.5
+
 - inspect now supports limit argument::
 
     myapp.control.inspect(limit=1).ping()
@@ -44,6 +46,14 @@
     The command must then support the interface of
     :class:`celery.bin.base.Command`.
 
+- contrib.migrate: New utilities to move tasks from one queue to another.
+
+    - :func:`~celery.contrib.migrate.move_tasks`
+    - :func:`~celery.contrib.migrate.move_task_by_id`
+
+- The task-sent event now contains ``exchange`` and ``routing_key``
+  fields.
+
 - Fixes bug with installing on Python 3.
 
     Fix contributed by Jed Smith.

celery/contrib/migrate.py

@@ -99,6 +99,27 @@ def migrate_tasks(source, dest, migrate=migrate_task, app=None,
 
 
 def move_tasks(conn, predicate, exchange, routing_key, app=None, **kwargs):
+    """Find tasks by filtering them and move the tasks to a new queue.
+
+    :param conn: Connection to use.
+    :param predicate: Filter function with signature ``(body, message)``.
+    :param exchange: Destination exchange.
+    :param routing_key: Destination routing key.
+
+    Also supports the same keyword arguments as :func:`start_filter`.
+
+    To demonstrate, the :func:`move_task_by_id` operation can be implemented
+    like this:
+
+    .. code-block:: python
+
+        def is_wanted_task(body, message):
+            if body['id'] == wanted_id:
+                return True
+
+        move_tasks(conn, is_wanted_task, exchange, routing_key)
+
+    """
     app = app_or_default(app)
     producer = app.amqp.TaskProducer(conn)
 
@@ -112,7 +133,16 @@ def move_tasks(conn, predicate, exchange, routing_key, app=None, **kwargs):
 
 
 def move_task_by_id(conn, task_id, exchange, routing_key, **kwargs):
+    """Find a task by id and move it to another queue.
+
+    :param conn: Connection to use.
+    :param task_id: Id of task to move.
+    :param exchange: Destination exchange.
+    :param exchange: Destination routing key.
+
+    Also supports the same keyword arguments as :func:`start_filter`.
 
+    """
     def predicate(body, message):
         if body['id'] == task_id:
             return True
@@ -128,6 +158,7 @@ def prepare_queues(queues):
                         for q in queues)
     if queues is None:
         queues = {}
+    return queues
 
 
 def start_filter(app, conn, filter, limit=None, timeout=1.0,

docs/userguide/monitoring.rst

@@ -545,7 +545,8 @@ This list contains the events sent by the worker, and their arguments.
 Task Events
 ~~~~~~~~~~~
 
-* ``task-sent(uuid, name, args, kwargs, retries, eta, expires, queue)``
+* ``task-sent(uuid, name, args, kwargs, retries, eta, expires,
+              queue, exchange, routing_key)``
 
    Sent when a task message is published and
    the :setting:`CELERY_SEND_TASK_SENT_EVENT` setting is enabled.