Improve documentation

celery/backends/cache.py

@@ -9,6 +9,7 @@ except ImportError:
 
 class Backend(BaseBackend):
     """Backend using the Django cache framework to store task metadata."""
+
     def __init__(self, *args, **kwargs):
         super(Backend, self).__init__(*args, **kwargs)
         self._cache = {}

celery/result.py

@@ -20,7 +20,11 @@ class BaseAsyncResult(object):
         self.backend = backend
 
     def is_done(self):
-        """Returns ``True`` if the task executed successfully."""
+        """Returns ``True`` if the task executed successfully.
+        
+        :rtype: bool
+        
+        """
         return self.backend.is_done(self.task_id)
 
     def get(self):
@@ -39,7 +43,11 @@ class BaseAsyncResult(object):
     def ready(self):
         """Returns ``True`` if the task executed successfully, or raised
         an exception. If the task is still pending, or is waiting for retry
-        then ``False`` is returned."""
+        then ``False`` is returned.
+        
+        :rtype: bool 
+
+        """
         status = self.backend.get_status(self.task_id)
         return status != "PENDING" or status != "RETRY"
 

celery/task.py

@@ -16,7 +16,16 @@ import pickle
 def delay_task(task_name, *args, **kwargs):
     """Delay a task for execution by the ``celery`` daemon.
 
-        >>> delay_task("update_record", name="George Constanza", age=32)
+        >>> r = delay_task("update_record", name="George Constanza", age=32)
+        >>> r.ready()
+        True
+        >>> r.result
+        "Record was updated"
+
+    Raises :class:`celery.registry.NotRegistered` exception if no such task 
+    has been registered in the task registry.
+
+    :rtype: :class:`celery.result.AsyncResult`. 
 
     """
     if task_name not in tasks:
@@ -38,6 +47,8 @@ def discard_all():
 
     Returns the number of tasks discarded.
 
+    :rtype: int
+
     """
     amqp_connection = DjangoAMQPConnection()
     consumer = TaskConsumer(connection=amqp_connection)
@@ -57,7 +68,11 @@ def mark_as_failure(task_id, exc):
 
 
 def is_done(task_id):
-    """Returns ``True`` if task with ``task_id`` has been executed."""
+    """Returns ``True`` if task with ``task_id`` has been executed.
+   
+    :rtype: bool
+
+    """
     return default_backend.is_done(task_id)
 
 
@@ -140,7 +155,11 @@ class Task(object):
 
     @classmethod
     def delay(cls, *args, **kwargs):
-        """Delay this task for execution by the ``celery`` daemon(s)."""
+        """Delay this task for execution by the ``celery`` daemon(s).
+        
+        :rtype: :class:`celery.result.AsyncResult`
+
+        """
         return delay_task(cls.name, *args, **kwargs)
 
 
@@ -380,6 +399,8 @@ def execute_remote(func, *args, **kwargs):
 
     The object must be picklable, so you can't use lambdas or functions
     defined in the REPL (the objects must have an associated module).
+
+    :rtype: :class:`celery.result.AsyncResult`
     
     """
     return ExecuteRemoteTask.delay(pickle.dumps(func), args, kwargs)

celery/timer.py

@@ -7,7 +7,21 @@ class TimeoutError(Exception):
 
 
 class EventTimer(object):
-    """Do something at an interval."""
+    """Do something at an interval.
+
+    .. attribute:: interval
+    
+        How often we call the event (in seconds).
+  
+    .. attribute:: event
+
+        The event callable to run every ``interval`` seconds.
+
+    .. attribute:: last_triggered
+
+        The last time, in unix timestamp format, the event was executed.
+    
+    """
 
     def __init__(self, event, interval=None):
         self.event = event
@@ -30,7 +44,18 @@ class EventTimer(object):
 
 
 class TimeoutTimer(object):
-    """A timer that raises ``TimeoutError`` when the time has run out."""
+    """A timer that raises :class:`TimeoutError` exception when the
+    time has run out.
+   
+    .. attribute:: timeout
+
+        The timeout in seconds.
+
+    .. attribute:: time_start
+
+        The time when the timeout timer instance was constructed.
+    
+    """
 
     def __init__(self, timeout):
         self.timeout = timeout

celery/worker.py

@@ -27,7 +27,15 @@ class UnknownTask(Exception):
 
 def jail(task_id, callable_, args, kwargs):
     """Wraps the task in a jail which saves the status and result
-    of the task execution to the task meta backend."""
+    of the task execution to the task meta backend.
+   
+    If the call results in an exception, it saves the exception as the task
+    result, and sets the task status to ``FAILURE``.
+
+    If the call was successful, it saves the result to the task result
+    backend, and sets the task status to ``DONE``.
+
+    """
     try:
         result = callable_(*args, **kwargs)
         mark_as_done(task_id, result)
@@ -39,6 +47,7 @@ def jail(task_id, callable_, args, kwargs):
 
 class TaskWrapper(object):
     """Class defining a task to be run."""
+
     def __init__(self, task_name, task_id, task_func, args, kwargs):
         self.task_name = task_name
         self.task_id = task_id
@@ -49,7 +58,14 @@ class TaskWrapper(object):
     @classmethod
     def from_message(cls, message):
         """Create a TaskWrapper from a message returned by
-        :class:`celery.messaging.TaskConsumer`."""
+        :class:`celery.messaging.TaskConsumer`.
+
+        If the message is not a proper task it raises
+        :class:`UnknownTask` exception.
+        
+        :rtype: :class:`TaskWrapper` instance.
+        
+        """
         message_data = simplejson.loads(message.body)
         task_name = message_data["task"]
         task_id = message_data["id"]
@@ -82,7 +98,10 @@ class TaskWrapper(object):
                         self.task_func, self.args, task_func_kwargs])
 
     def execute_using_pool(self, pool, loglevel, logfile):
-        """Like ``execute``, but using the ``multiprocessing`` pool."""
+        """Like ``execute``, but using the ``multiprocessing`` pool.
+       
+        :rtype: :class:`multiprocessing.AsyncResult` instance.
+        """
         task_func_kwargs = self.extend_kwargs_with_logging(loglevel, logfile)
         return pool.apply_async(jail, [self.task_id, self.task_func,
                                        self.args, task_func_kwargs])
@@ -91,7 +110,43 @@ class TaskWrapper(object):
 class TaskDaemon(object):
     """Executes tasks waiting in the task queue.
 
-    ``concurrency`` is the number of simultaneous processes.
+    .. attribute:: concurrency
+
+        The number of simultaneous processes doing work (default:
+        ``celery.conf.DAEMON_CONCURRENCY``)
+
+    .. attribute:: loglevel
+
+        The loglevel used (default: ``logging.INFO``)
+
+    .. attribute:: logfile
+
+        The logfile used, if no logfile is specified it uses ``stderr``
+        (default: ``celery.conf.DAEMON_LOG_FILE``).
+
+    .. attribute:: queue_wakeup_after
+
+        The time it takes for the daemon to wake up after the queue is empty,
+        so it can check for more work
+        (default: ``celery.conf.QUEUE_WAKEUP_AFTER``).
+
+    .. attribute:: empty_msg_emit_every
+
+        How often the daemon emits the ``Waiting for queue`` message.
+        If this is ``None``, the message will never be logged.
+        (default: ``celery.conf.EMPTY_MSG_EMIT_EVERY``)
+
+    .. attribute:: logger
+
+        The :class:`logging.Logger` instance used for logging.
+
+    .. attribute:: pool
+        The :class:`multiprocessing.Pool` instance used.
+
+    .. attribute:: task_consumer
+
+        The :class:`celery.messaging.TaskConsumer` instance used.
+
     """
     loglevel = logging.ERROR
     concurrency = DAEMON_CONCURRENCY
@@ -112,18 +167,29 @@ class TaskDaemon(object):
         self.reset_connection()
 
     def reset_connection(self):
+        """Reset the AMQP connection, and reinitialize the
+        :class:`celery.messaging.TaskConsumer` instance."""
         if self.task_consumer:
             self.task_consumer.connection.close()
         amqp_connection = DjangoAMQPConnection()
         self.task_consumer = TaskConsumer(connection=amqp_connection)
 
     def connection_diagnostics(self):
+        """Diagnose the AMQP connection, and reset if necessary."""
         if not self.task_consumer.channel.connection:
             self.logger.info(
                     "AMQP Connection has died, restoring connection.")
             self.reset_connection()
 
     def receive_message(self):
+        """Receive the next message from the Task consumer queue.
+       
+        Tries to reset the AMQP connection if not available.
+        Returns ``None`` if no message is waiting on the queue.
+
+        :rtype: :class:`carrot.messaging.Message` instance.
+        
+        """
         self.connection_diagnostics()
         message = self.task_consumer.fetch()
         if message is not None:
@@ -131,6 +197,7 @@ class TaskDaemon(object):
         return message
 
     def receive_message_cc(self):
+        """UNUSED."""
         amqp_connection = DjangoAMQPConnection()
         task_consumer = TaskConsumer(connection=amqp_connection)
         message = task_consumer.fetch()
@@ -140,6 +207,13 @@ class TaskDaemon(object):
         return message
 
     def fetch_next_task(self):
+        """Fetch the next task from the AMQP broker.
+       
+        Raises :class`EmptyQueue` exception if there is no messages
+        waiting on the queue.
+
+        :rtype: :class:`TaskWrapper` instance.
+        """
         message = self.receive_message()
         if message is None: # No messages waiting.
             raise EmptyQueue()
@@ -151,6 +225,12 @@ class TaskDaemon(object):
         return task, message
 
     def execute_next_task(self):
+        """Execute the next task on the queue using the multiprocessing
+        pool.
+        
+        Catches all exceptions and logs them with level ``logging.CRITICAL``. 
+
+        """
         task, message = self.fetch_next_task()
 
         try:
@@ -166,7 +246,7 @@ class TaskDaemon(object):
     def run_periodic_tasks(self):
         """Schedule all waiting periodic tasks for execution.
        
-        Returns list of :class:`celery.models.PeriodicTaskMeta` objects.
+        :rtype: list of :class:`celery.models.PeriodicTaskMeta` objects.
         """
         waiting_tasks = PeriodicTaskMeta.objects.get_waiting_tasks()
         [waiting_task.delay()