Merged from upstream/master

Changelog

@@ -9,9 +9,29 @@
 
 2.2.4
 =====
-:release-date: 2011-02-14 11:00 AM CET
+:release-date: 2011-02-19 12:00 AM CET
+
+* celeryd: 2.2.3 broke error logging, resulting in tracebacks not being logged.
+
+* AMQP result backend: Polling task states did not work properly if there were
+  more than one result message in the queue.
+
+* ``TaskSet.apply_async()`` and ``TaskSet.apply()`` now supports an optional
+  ``taskset_id`` keyword argument (Issue #331).
+
+* The current taskset id (if any) is now available in the task context as
+  ``request.context`` (Issue #329).
+
+* SQLAlchemy result backend: `date_done` was no longer part of the results as it had
+  been accidentally removed.  It is now available again (Issue #325).
+
+* SQLAlchemy result backend: Added unique constraint on `Task.task_id` and
+  `TaskSet.taskset_id`.  Tables needs to be recreated for this to take effect.
+
+* Fixed exception raised when iterating on the result of ``TaskSet.apply()``.
+
+* Tasks Userguide: Added section on choosing a result backend.
 
-* celeryd: 2.2.3 broke error logging, so tracebacks would no longer be logged.
 
 .. _v224-fixes:
 

README.rst

@@ -4,7 +4,7 @@
 
 .. image:: http://cloud.github.com/downloads/ask/celery/celery_favicon_128.png
 
-:Version: 2.2.3
+:Version: 2.2.4
 :Web: http://celeryproject.org/
 :Download: http://pypi.python.org/pypi/celery/
 :Source: http://github.com/ask/celery/

celery/db/models.py

@@ -33,7 +33,8 @@ class Task(ResultModelBase):
         return {"task_id": self.task_id,
                 "status": self.status,
                 "result": self.result,
-                "traceback": self.traceback}
+                "traceback": self.traceback,
+                "date_done": self.date_done}
 
     def __repr__(self):
         return "<Task %s state: %s>" % (self.task_id, self.status)
@@ -57,7 +58,8 @@ class TaskSet(ResultModelBase):
 
     def to_dict(self):
         return {"taskset_id": self.taskset_id,
-                "result": self.result}
+                "result": self.result,
+                "date_done": self.date_done}
 
     def __repr__(self):
         return u"<TaskSet: %s>" % (self.taskset_id, )

celery/log.py

@@ -42,9 +42,9 @@ class ColorFormatter(logging.Formatter):
         if self.use_color and color:
             try:
                 record.msg = color(safe_str(record.msg))
-            except Exception, exc:
-                record.msg = "<Unrepresentable %r: %r>" % (type(record.msg),
-                                                           traceback.format_stack())
+            except Exception:
+                record.msg = "<Unrepresentable %r: %r>" % (
+                        type(record.msg), traceback.format_stack())
 
         # Very ugly, but have to make sure processName is supported
         # by foreign logger instances.

celery/signals.py

@@ -3,8 +3,8 @@
 celery.signals
 ==============
 
-Signals allows decoupled applications to receive notifications when certain actions
-occur.
+Signals allows decoupled applications to receive notifications when
+certain actions occur elsewhere in the application.
 
 :copyright: (c) 2009 - 2011 by Ask Solem.
 :license: BSD, see LICENSE for more details.

celery/tests/test_backends/test_amqp.py

@@ -2,6 +2,7 @@ import socket
 import sys
 
 from datetime import timedelta
+from Queue import Empty, Queue
 
 from celery import states
 from celery.app import app_or_default
@@ -123,8 +124,15 @@ class test_AMQPBackend(unittest.TestCase):
 
     def test_poll_result(self):
 
+        results = Queue()
+
+        class Message(object):
+
+            def __init__(self, **merge):
+                self.payload = dict({"status": states.STARTED,
+                                     "result": None}, **merge)
+
         class MockBinding(object):
-            get_returns = [True]
 
             def __init__(self, *args, **kwargs):
                 pass
@@ -136,23 +144,37 @@ class test_AMQPBackend(unittest.TestCase):
                 pass
 
             def get(self, no_ack=False):
-                if self.get_returns[0]:
-                    class Object(object):
-                        payload = {"status": "STARTED",
-                                "result": None}
-                    return Object()
+                try:
+                    return results.get(block=False)
+                except Empty:
+                    pass
 
         class MockBackend(AMQPBackend):
             Queue = MockBinding
 
         backend = MockBackend()
-        backend.poll(gen_unique_id())
+
+        # FFWD's to the latest state.
+        results.put(Message(status=states.RECEIVED, seq=1))
+        results.put(Message(status=states.STARTED, seq=2))
+        results.put(Message(status=states.FAILURE, seq=3))
+        r1 = backend.poll(gen_unique_id())
+        self.assertDictContainsSubset({"status": states.FAILURE,
+                                       "seq": 3}, r1,
+                                       "FFWDs to the last state")
+
+        # Caches last known state.
+        results.put(Message())
         uuid = gen_unique_id()
         backend.poll(uuid)
-        self.assertIn(uuid, backend._cache)
-        MockBinding.get_returns[0] = False
+        self.assertIn(uuid, backend._cache, "Caches last known state")
+
+        # Returns cache if no new states.
+        results.queue.clear()
+        assert not results.qsize()
         backend._cache[uuid] = "hello"
-        self.assertEqual(backend.poll(uuid), "hello")
+        self.assertEqual(backend.poll(uuid), "hello",
+                         "Returns cache if no new states")
 
     def test_wait_for(self):
         b = self.create_backend()

celery/tests/test_task/test_task.py

@@ -438,7 +438,7 @@ class TestTaskSet(unittest.TestCase):
     def test_named_taskset(self):
         prefix = "test_named_taskset-"
         ts = task.TaskSet([return_True_task.subtask([1])])
-        res = ts.apply(taskset_id=prefix+gen_unique_id())
+        res = ts.apply(taskset_id=prefix + gen_unique_id())
         self.assertTrue(res.taskset_id.startswith(prefix))
 
 class TestTaskApply(unittest.TestCase):