浏览代码

Better, greater documentation

Ask Solem 16 年之前
父节点
当前提交
59e01ea8c4

+ 22 - 1
celery/backends/__init__.py

@@ -10,7 +10,7 @@ CELERY_BACKEND = getattr(settings, "CELERY_BACKEND", DEFAULT_BACKEND)
 def get_backend_cls(backend):
 def get_backend_cls(backend):
     """Get backend class by name.
     """Get backend class by name.
 
 
-    If the name does not include ``.`` (is not fully qualified),
+    If the name does not include "``.``" (is not fully qualified),
     ``celery.backends.`` will be prepended to the name. e.g.
     ``celery.backends.`` will be prepended to the name. e.g.
     ``database`` becomes ``celery.backends.database``.
     ``database`` becomes ``celery.backends.database``.
 
 
@@ -21,6 +21,27 @@ def get_backend_cls(backend):
     backend_module = sys.modules[backend]
     backend_module = sys.modules[backend]
     return getattr(backend_module, 'Backend')
     return getattr(backend_module, 'Backend')
 
 
+"""
+.. function:: get_default_backend_cls()
+
+    Get the backend class specified in :settings:`CELERY_BACKEND`.
+
+"""
 get_default_backend_cls = partial(get_backend_cls, CELERY_BACKEND)
 get_default_backend_cls = partial(get_backend_cls, CELERY_BACKEND)
+
+
+"""
+.. class:: DefaultBackend
+
+    The backend class specified in :setting:`CELERY_BACKEND`.
+
+"""
 DefaultBackend = get_default_backend_cls()
 DefaultBackend = get_default_backend_cls()
+
+"""
+.. data:: default_backend
+
+    An instance of :class:`DefaultBackend`.
+
+"""
 default_backend = DefaultBackend()
 default_backend = DefaultBackend()

+ 4 - 4
celery/backends/base.py

@@ -51,11 +51,11 @@ class BaseBackend(object):
         """Wait for task and return its result.
         """Wait for task and return its result.
 
 
         If the task raises an exception, this exception
         If the task raises an exception, this exception
-        will be re-raised by ``wait_for``.
+        will be re-raised by :func:`wait_for`.
 
 
-        If ``timeout`` is not ``None``, this raises
-        ``celery.timer.TimeoutError`` if the operation takes longer than
-        ``timeout`` seconds.
+        If ``timeout`` is not ``None``, this raises the
+        :class:`celery.timer.TimeoutError` exception if the operation takes
+        longer than ``timeout`` seconds.
 
 
         """
         """
         timeout_timer = TimeoutTimer(timeout)
         timeout_timer = TimeoutTimer(timeout)

+ 6 - 0
celery/backends/tyrant.py

@@ -33,6 +33,12 @@ class Backend(BaseBackend):
     tyrant_port = None
     tyrant_port = None
 
 
     def __init__(self, tyrant_host=None, tyrant_port=None):
     def __init__(self, tyrant_host=None, tyrant_port=None):
+        """Initialize Tokyo Tyrant backend instance.
+
+        Raises :class:`django.core.exceptions.ImproperlyConfigured` if
+        :setting:`TT_HOST` or :setting:`TT_PORT` is not set.
+
+        """
         self.tyrant_host = kwargs.get("tyrant_host", 
         self.tyrant_host = kwargs.get("tyrant_host", 
                             getattr(settings, "TT_HOST", self.tyrant_host))
                             getattr(settings, "TT_HOST", self.tyrant_host))
         self.tyrant_port = kwargs.get("tyrant_port",
         self.tyrant_port = kwargs.get("tyrant_port",

+ 53 - 6
celery/datastructures.py

@@ -1,9 +1,20 @@
-"""celery.datastructures"""
+"""
+
+Custom Datastructures
+
+"""
+
 from UserList import UserList
 from UserList import UserList
 
 
 
 
 class PositionQueue(UserList):
 class PositionQueue(UserList):
-    """A positional queue with filled/unfilled slots."""
+    """A positional queue of a specific length, with slots that are either
+    filled or unfilled. When all of the positions are filled, the queue
+    is considered :meth:`full`.
+   
+    :param length: The number of items required for the queue to be filled.
+
+    """
 
 
     class UnfilledPosition(object):
     class UnfilledPosition(object):
         """Describes an unfilled slot."""
         """Describes an unfilled slot."""
@@ -11,16 +22,15 @@ class PositionQueue(UserList):
             self.position = position
             self.position = position
 
 
     def __init__(self, length):
     def __init__(self, length):
-        """Initialize a position queue with ``length`` slots."""
         self.length = length
         self.length = length
         self.data = map(self.UnfilledPosition, xrange(length))
         self.data = map(self.UnfilledPosition, xrange(length))
 
 
     def full(self):
     def full(self):
-        """Returns ``True`` if all the positions has been filled."""
+        """Returns ``True`` if all of the slots has been filled."""
         return len(self) >= self.length
         return len(self) >= self.length
 
 
     def __len__(self):
     def __len__(self):
-        """len(self) -> number of positions filled with real values."""
+        """``len(self)`` -> number of slots filled with real values."""
         return len(self.filled)
         return len(self.filled)
 
 
     @property
     @property
@@ -32,7 +42,30 @@ class PositionQueue(UserList):
         
         
 class TaskProcessQueue(UserList):
 class TaskProcessQueue(UserList):
     """Queue of running child processes, which starts waiting for the
     """Queue of running child processes, which starts waiting for the
-    processes to finish when the queue limit is reached."""
+    processes to finish when the queue limit is reached.
+
+    :param limit: see :attr:`limit` attribute.
+
+    :param logger: see :attr:`logger` attribute.
+
+    :param done_msg: see :attr:`done_msg` attribute.
+
+
+    .. attribute:: limit
+
+        The number of processes that can run simultaneously until
+        we start collecting results.
+
+    .. attribute:: logger
+
+        The logger used to print the :attr:`done_msg`.
+
+    .. attribute:: done_msg
+
+        Message logged when a tasks result has been collected.
+        The message is logged with loglevel :const:`logging.INFO`.
+
+    """
 
 
     def __init__(self, limit, logger=None, done_msg=None):
     def __init__(self, limit, logger=None, done_msg=None):
         self.limit = limit
         self.limit = limit
@@ -41,6 +74,20 @@ class TaskProcessQueue(UserList):
         self.data = []
         self.data = []
         
         
     def add(self, result, task_name, task_id):
     def add(self, result, task_name, task_id):
+        """Add a process to the queue.
+        
+        If the queue is full, it will start to collect return values from
+        the tasks executed. When all return values has been collected,
+        it deletes the current queue and is ready to accept new processes.
+
+        :param result: A :class:`multiprocessing.AsyncResult` instance, as
+            returned by :meth:`multiprocessing.Pool.apply_async`.
+
+        :param task_name: Name of the task executed.
+
+        :param task_id: Id of the task executed.
+        
+        """
         self.data.append([result, task_name, task_id])
         self.data.append([result, task_name, task_id])
 
 
         if self.data and len(self.data) >= self.limit:
         if self.data and len(self.data) >= self.limit:

+ 2 - 2
celery/managers.py

@@ -5,7 +5,7 @@ from datetime import datetime, timedelta
 
 
 
 
 class TaskManager(models.Manager):
 class TaskManager(models.Manager):
-    """Manager for ``Task`` models."""
+    """Manager for :class:`celery.models.Task` models."""
     
     
     def get_task(self, task_id):
     def get_task(self, task_id):
         """Get task meta for task by ``task_id``."""
         """Get task meta for task by ``task_id``."""
@@ -36,7 +36,7 @@ class TaskManager(models.Manager):
 
 
 
 
 class PeriodicTaskManager(models.Manager):
 class PeriodicTaskManager(models.Manager):
-    """Manager for ``PeriodicTask`` models."""
+    """Manager for :class:`celery.models.PeriodicTask` models."""
 
 
     def get_waiting_tasks(self):
     def get_waiting_tasks(self):
         """Get all waiting periodic tasks."""
         """Get all waiting periodic tasks."""

+ 7 - 3
celery/messaging.py

@@ -1,4 +1,8 @@
-"""celery.messaging"""
+"""
+
+Sending and Receiving Messages
+
+"""
 from carrot.messaging import Publisher, Consumer
 from carrot.messaging import Publisher, Consumer
 from celery import conf
 from celery import conf
 import uuid
 import uuid
@@ -6,11 +10,11 @@ import uuid
 
 
 class NoProcessConsumer(Consumer):
 class NoProcessConsumer(Consumer):
     """A consumer that raises an error if used with wait callbacks (i.e.
     """A consumer that raises an error if used with wait callbacks (i.e.
-    it doesn't support ``carrot.messaging.Consumer.wait``)."""
+    it doesn't support :meth:`carrot.messaging.Consumer.wait``)."""
     
     
     def receive(self, message_data, message):
     def receive(self, message_data, message):
         raise NotImplementedError(
         raise NotImplementedError(
-                "Don't use process_next() or wait() with the TaskConsumer!")
+                "This consumer doesn't support process_next() or wait()")
 
 
 
 
 class TaskPublisher(Publisher):
 class TaskPublisher(Publisher):

+ 0 - 1
celery/models.py

@@ -1,4 +1,3 @@
-"""celery.models"""
 from django.db import models
 from django.db import models
 from celery.registry import tasks
 from celery.registry import tasks
 from celery.managers import TaskManager, PeriodicTaskManager
 from celery.managers import TaskManager, PeriodicTaskManager

+ 17 - 6
celery/registry.py

@@ -1,4 +1,3 @@
-"""celery.registry"""
 from celery import discovery
 from celery import discovery
 from UserDict import UserDict
 from UserDict import UserDict
 
 
@@ -21,7 +20,7 @@ class TaskRegistry(UserDict):
         self.data = {}
         self.data = {}
 
 
     def autodiscover(self):
     def autodiscover(self):
-        """Autodiscover tasks using ``celery.discovery.autodiscover``."""
+        """Autodiscovers tasks using :func:`celery.discovery.autodiscover`."""
         discovery.autodiscover()
         discovery.autodiscover()
 
 
     def register(self, task, task_name=None):
     def register(self, task, task_name=None):
@@ -30,8 +29,9 @@ class TaskRegistry(UserDict):
         Task can either be a regular function, or a class inheriting
         Task can either be a regular function, or a class inheriting
         from :class:`celery.task.Task`.
         from :class:`celery.task.Task`.
 
 
-        If the task is a regular function, the ``task_name`` argument is
-        required.
+        :keyword task_name: Required if the task is a regular function.
+
+        :raises AlreadyRegistered: if the task is already registered.
         
         
         """
         """
         is_class = False
         is_class = False
@@ -49,7 +49,13 @@ class TaskRegistry(UserDict):
             self.data[task_name] = task
             self.data[task_name] = task
 
 
     def unregister(self, task_name):
     def unregister(self, task_name):
-        """Unregister task by name."""
+        """Unregister task by name.
+       
+        :param task_name: name of the task to unregister.
+
+        :raises NotRegistered: if the task has not been registered.
+
+        """
         if hasattr(task_name, "run"):
         if hasattr(task_name, "run"):
             task_name = task_name.name
             task_name = task_name.name
         if task_name not in self.data:
         if task_name not in self.data:
@@ -79,5 +85,10 @@ class TaskRegistry(UserDict):
         """Get task by name."""
         """Get task by name."""
         return self.data[task_name]
         return self.data[task_name]
 
 
-"""This is the global task registry."""
+"""
+.. data:: tasks
+
+    The global task registry.
+
+"""
 tasks = TaskRegistry()
 tasks = TaskRegistry()

+ 56 - 12
celery/result.py

@@ -1,4 +1,8 @@
-"""celery.result"""
+"""
+
+Asynchronous result types.
+
+"""
 from celery.backends import default_backend
 from celery.backends import default_backend
 
 
 
 
@@ -16,6 +20,13 @@ class BaseAsyncResult(object):
     """
     """
 
 
     def __init__(self, task_id, backend):
     def __init__(self, task_id, backend):
+        """Create a result instance.
+
+        :param task_id: id of the task this is a result for.
+
+        :param backend: task result backend used.
+
+        """
         self.task_id = task_id
         self.task_id = task_id
         self.backend = backend
         self.backend = backend
 
 
@@ -28,16 +39,22 @@ class BaseAsyncResult(object):
         return self.backend.is_done(self.task_id)
         return self.backend.is_done(self.task_id)
 
 
     def get(self):
     def get(self):
-        """Alias to ``wait``."""
+        """Alias to :func:`wait`."""
         return self.wait()
         return self.wait()
 
 
     def wait(self, timeout=None):
     def wait(self, timeout=None):
-        """Return the result when it arrives.
+        """Wait for task, and return the result when it arrives.
+       
+        :keyword timeout: How long to wait in seconds, before the
+            operation times out.
+        
+        :raises celery.timer.TimeoutError: if ``timeout`` is not ``None`` and
+            the result does not arrive within ``timeout`` seconds.
+        
+        If the remote call raised an exception then that
+        exception will be re-raised.
         
         
-        If timeout is not ``None`` and the result does not arrive within
-        ``timeout`` seconds then ``celery.backends.base.TimeoutError`` is
-        raised. If the remote call raised an exception then that exception
-        will be reraised by get()."""
+        """
         return self.backend.wait_for(self.task_id, timeout=timeout)
         return self.backend.wait_for(self.task_id, timeout=timeout)
 
 
     def ready(self):
     def ready(self):
@@ -52,11 +69,11 @@ class BaseAsyncResult(object):
         return status != "PENDING" or status != "RETRY"
         return status != "PENDING" or status != "RETRY"
 
 
     def successful(self):
     def successful(self):
-        """Alias to ``is_done``."""
+        """Alias to :func:`is_done`."""
         return self.is_done()
         return self.is_done()
 
 
     def __str__(self):
     def __str__(self):
-        """str(self) -> self.task_id"""
+        """``str(self)`` -> ``self.task_id``"""
         return self.task_id
         return self.task_id
 
 
     def __repr__(self):
     def __repr__(self):
@@ -64,14 +81,41 @@ class BaseAsyncResult(object):
 
 
     @property
     @property
     def result(self):
     def result(self):
-        """The tasks resulting value."""
+        """When the task is executed, this contains the return value.
+       
+        If the task resulted in failure, this will be the exception instance
+        raised.
+        """
         if self.status == "DONE" or self.status == "FAILURE":
         if self.status == "DONE" or self.status == "FAILURE":
             return self.backend.get_result(self.task_id)
             return self.backend.get_result(self.task_id)
         return None
         return None
 
 
     @property
     @property
     def status(self):
     def status(self):
-        """The current status of the task."""
+        """The current status of the task.
+       
+        Can be one of the following:
+
+            *PENDING*
+
+                The task is waiting for execution.
+
+            *RETRY*
+
+                The task is to be retried, possibly because of failure.
+
+            *FAILURE*
+
+                The task raised an exception, or has been retried more times
+                than its limit. The :attr:`result` attribute contains the
+                exception raised.
+
+            *DONE*
+
+                The task executed successfully. The :attr:`result` attribute
+                contains the resulting value.
+
+        """
         return self.backend.get_status(self.task_id)
         return self.backend.get_status(self.task_id)
 
 
 
 
@@ -84,7 +128,7 @@ class AsyncResult(BaseAsyncResult):
 
 
     .. attribute:: backend
     .. attribute:: backend
     
     
-        Instance of ``celery.backends.DefaultBackend``.
+        Instance of :class:`celery.backends.DefaultBackend`.
 
 
     """
     """
     def __init__(self, task_id):
     def __init__(self, task_id):

+ 191 - 54
celery/task.py

@@ -1,3 +1,8 @@
+"""
+
+Working with tasks and task sets.
+
+"""
 from carrot.connection import DjangoAMQPConnection
 from carrot.connection import DjangoAMQPConnection
 from celery.log import setup_logger
 from celery.log import setup_logger
 from celery.registry import tasks
 from celery.registry import tasks
@@ -16,17 +21,25 @@ import pickle
 def delay_task(task_name, *args, **kwargs):
 def delay_task(task_name, *args, **kwargs):
     """Delay a task for execution by the ``celery`` daemon.
     """Delay a task for execution by the ``celery`` daemon.
 
 
+    :param task_name: the name of a task registered in the task registry.
+
+    :param \*args: positional arguments to pass on to the task.
+    
+    :param \*\*kwargs: keyword arguments to pass on to the task.
+    
+    :raises celery.registry.NotRegistered: exception if no such task 
+        has been registered in the task registry.
+
+    :rtype: :class:`celery.result.AsyncResult`. 
+
+    Example
+
         >>> r = delay_task("update_record", name="George Constanza", age=32)
         >>> r = delay_task("update_record", name="George Constanza", age=32)
         >>> r.ready()
         >>> r.ready()
         True
         True
         >>> r.result
         >>> r.result
         "Record was updated"
         "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:
     if task_name not in tasks:
         raise tasks.NotRegistered(
         raise tasks.NotRegistered(
@@ -45,7 +58,7 @@ def discard_all():
     This will ignore all tasks waiting for execution, and they will
     This will ignore all tasks waiting for execution, and they will
     be deleted from the messaging server.
     be deleted from the messaging server.
 
 
-    Returns the number of tasks discarded.
+    :returns: the number of tasks discarded.
 
 
     :rtype: int
     :rtype: int
 
 
@@ -58,13 +71,25 @@ def discard_all():
 
 
 
 
 def mark_as_done(task_id, result):
 def mark_as_done(task_id, result):
-    """Mark task as done (executed)."""
-    return default_backend.mark_as_done(task_id, result)
+    """Mark task as done (executed successfully).
+   
+    :param task_id: id of the task.
+    
+    :param result: the return value of the task.
+
+    """
+    default_backend.mark_as_done(task_id, result)
 
 
 
 
 def mark_as_failure(task_id, exc):
 def mark_as_failure(task_id, exc):
-    """Mark task as done (executed)."""
-    return default_backend.mark_as_failure(task_id, exc)
+    """Mark task as done (executed).
+    
+    :param task_id: id of the task.
+
+    :param exc: the exception instance raised by the task.
+
+    """
+    default_backend.mark_as_failure(task_id, exc)
 
 
 
 
 def is_done(task_id):
 def is_done(task_id):
@@ -79,10 +104,29 @@ def is_done(task_id):
 class Task(object):
 class Task(object):
     """A task that can be delayed for execution by the ``celery`` daemon.
     """A task that can be delayed for execution by the ``celery`` daemon.
 
 
-    All subclasses of ``Task`` has to define the ``name`` attribute, which is
-    the name of the task that can be passed to ``celery.task.delay_task``,
-    it also has to define the ``run`` method, which is the actual method the
-    ``celery`` daemon executes.
+    All subclasses of :class:`Task` must define the :meth:`run` method,
+    which is the actual method the ``celery`` daemon executes.
+
+    The :meth:`run` method supports both positional, and keyword arguments.
+    
+    .. attribute:: name
+
+        *REQUIRED* All subclasses of :class:`Task` has to define the
+        :attr:`name` attribute. This is the name of the task, registered
+        in the task registry, and passed to :func:`delay_task`.
+        
+    .. attribute:: type
+
+        The type of task, currently this can be ``regular``, or ``periodic``,
+        however if you want a periodic task, you should subclass
+        :class:`PeriodicTask` instead.
+        
+    :raises NotImplementedError: if the :attr:`name` attribute is not set.
+
+    The resulting class is callable, which if called will apply the
+    :meth:`run` method.
+
+    Examples
     
     
     This is a simple task just logging a message,
     This is a simple task just logging a message,
 
 
@@ -97,7 +141,7 @@ class Task(object):
         ...         return 42
         ...         return 42
         ... tasks.register(MyTask)
         ... tasks.register(MyTask)
 
 
-    You can delay the task using the classmethod ``delay``...
+    You can delay the task using the classmethod :meth:`delay`...
 
 
         >>> result = MyTask.delay(some_arg="foo")
         >>> result = MyTask.delay(some_arg="foo")
         >>> result.status # after some time
         >>> result.status # after some time
@@ -105,11 +149,12 @@ class Task(object):
         >>> result.result
         >>> result.result
         42
         42
 
 
-    ...or using the ``celery.task.delay_task`` function, by passing the
-    name of the task.
+    ...or using the :func:`delay_task` function, by passing the name of
+    the task.
 
 
         >>> from celery.task import delay_task
         >>> from celery.task import delay_task
-        >>> delay_task(MyTask.name, some_arg="foo")
+        >>> result = delay_task(MyTask.name, some_arg="foo")
+
 
 
     """
     """
     name = None
     name = None
@@ -123,26 +168,55 @@ class Task(object):
             raise NotImplementedError("Tasks must define a name attribute.")
             raise NotImplementedError("Tasks must define a name attribute.")
 
 
     def __call__(self, *args, **kwargs):
     def __call__(self, *args, **kwargs):
-        """The ``__call__`` is called when you do ``Task().run()`` and calls
-        the ``run`` method. It also catches any exceptions and logs them."""
         return self.run(*args, **kwargs)
         return self.run(*args, **kwargs)
 
 
     def run(self, *args, **kwargs):
     def run(self, *args, **kwargs):
-        """The actual task. All subclasses of :class:`Task` must define
-        the run method, if not a ``NotImplementedError`` exception is raised.
+        """*REQUIRED* The actual task.
+        
+        All subclasses of :class:`Task` must define the run method.
+        
+        :raises NotImplementedError: by default, so you have to override
+            this method in your subclass.
+
         """
         """
         raise NotImplementedError("Tasks must define a run method.")
         raise NotImplementedError("Tasks must define a run method.")
 
 
     def get_logger(self, **kwargs):
     def get_logger(self, **kwargs):
-        """Get a process-aware logger object."""
+        """Get process-aware logger object.
+
+        See :func:`celery.log.setup_logger`.
+        
+        """
         return setup_logger(**kwargs)
         return setup_logger(**kwargs)
 
 
     def get_publisher(self):
     def get_publisher(self):
-        """Get a celery task message publisher."""
+        """Get a celery task message publisher.
+        
+        :rtype: :class:`celery.messaging.TaskPublisher`.
+
+        Please be sure to close the AMQP connection when you're done
+        with this object, i.e:
+
+            >>> publisher = self.get_publisher()
+            >>> # do something with publisher
+            >>> publisher.connection.close()
+
+        """
         return TaskPublisher(connection=DjangoAMQPConnection())
         return TaskPublisher(connection=DjangoAMQPConnection())
 
 
     def get_consumer(self):
     def get_consumer(self):
-        """Get a celery task message consumer."""
+        """Get a celery task message consumer.
+       
+        :rtype: :class:`celery.messaging.TaskConsumer`.
+
+        Please be sure to close the AMQP connection when you're done
+        with this object. i.e:
+
+            >>> consumer = self.get_consumer()
+            >>> # do something with consumer
+            >>> consumer.connection.close()
+
+        """
         return TaskConsumer(connection=DjangoAMQPConnection())
         return TaskConsumer(connection=DjangoAMQPConnection())
 
 
     def requeue(self, task_id, args, kwargs):
     def requeue(self, task_id, args, kwargs):
@@ -156,33 +230,57 @@ class Task(object):
     @classmethod
     @classmethod
     def delay(cls, *args, **kwargs):
     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).
+
+        :param \*args: positional arguments passed on to the task.
+
+        :param \*\*kwargs: keyword arguments passed on to the task.
         
         
         :rtype: :class:`celery.result.AsyncResult`
         :rtype: :class:`celery.result.AsyncResult`
 
 
+        See :func:`delay_task`.
+
         """
         """
         return delay_task(cls.name, *args, **kwargs)
         return delay_task(cls.name, *args, **kwargs)
 
 
 
 
 class TaskSet(object):
 class TaskSet(object):
     """A task containing several subtasks, making it possible
     """A task containing several subtasks, making it possible
-    to track how many, or when all of the tasks are completed.
-    
+    to track how many, or when all of the tasks has been completed.
+
+    :param task: The task class or name.
+        Can either be a fully qualified task name, or a task class.
+
+    :param args: A list of args, kwargs pairs.
+        e.g. ``[[args1, kwargs1], [args2, kwargs2], ..., [argsN, kwargsN]]``
+
+
+    .. attribute:: task_name
+
+        The name of the task.
+
+    .. attribute:: arguments
+
+        The arguments, as passed to the task set constructor.
+
+    .. attribute:: total
+
+        Total number of tasks in this task set.
+   
+    Example
+
         >>> from djangofeeds.tasks import RefreshFeedTask
         >>> from djangofeeds.tasks import RefreshFeedTask
         >>> taskset = TaskSet(RefreshFeedTask, args=[
         >>> taskset = TaskSet(RefreshFeedTask, args=[
-        ...                 {"feed_url": "http://cnn.com/rss"},
-        ...                 {"feed_url": "http://bbc.com/rss"},
-        ...                 {"feed_url": "http://xkcd.com/rss"}])
+        ...                 [], {"feed_url": "http://cnn.com/rss"},
+        ...                 [], {"feed_url": "http://bbc.com/rss"},
+        ...                 [], {"feed_url": "http://xkcd.com/rss"}])
 
 
         >>> taskset_id, subtask_ids = taskset.run()
         >>> taskset_id, subtask_ids = taskset.run()
+        >>> list_of_return_values = taskset.join()
         
         
 
 
     """
     """
 
 
     def __init__(self, task, args):
     def __init__(self, task, args):
-        """``task`` can be either a fully qualified task name, or a task
-        class, args is a list of arguments for the subtasks.
-        """
-
         try:
         try:
             task_name = task.name
             task_name = task.name
         except AttributeError:
         except AttributeError:
@@ -195,7 +293,12 @@ class TaskSet(object):
     def run(self):
     def run(self):
         """Run all tasks in the taskset.
         """Run all tasks in the taskset.
 
 
-        Returns a tuple with the taskset id, and a list of subtask id's.
+        :returns: A tuple containing the taskset id, and a list
+            of subtask ids.
+
+        :rtype: tuple
+
+        Example
 
 
             >>> ts = RefreshFeeds([
             >>> ts = RefreshFeeds([
             ...         ["http://foo.com/rss", {}],
             ...         ["http://foo.com/rss", {}],
@@ -225,10 +328,11 @@ class TaskSet(object):
         return taskset_id, subtask_ids
         return taskset_id, subtask_ids
 
 
     def iterate(self):
     def iterate(self):
-        """Iterate over the results returned after calling ``run()``.
+        """Iterate over the results returned after calling :meth:`run`.
         
         
         If any of the tasks raises an exception, the exception will
         If any of the tasks raises an exception, the exception will
-        be reraised by ``iterate``.
+        be re-raised.
+
         """
         """
         taskset_id, subtask_ids = self.run()
         taskset_id, subtask_ids = self.run()
         results = dict([(task_id, AsyncResult(task_id))
         results = dict([(task_id, AsyncResult(task_id))
@@ -246,12 +350,16 @@ class TaskSet(object):
         and return a list with them ordered by the order of which they
         and return a list with them ordered by the order of which they
         were called.
         were called.
 
 
+        :keyword timeout: The time in seconds, how long
+            it will wait for results, before the operation times out.
+
+        :raises celery.timer.TimeoutError: if ``timeout`` is not ``None``
+            and the operation takes longer than ``timeout`` seconds.
+
         If any of the tasks raises an exception, the exception
         If any of the tasks raises an exception, the exception
-        will be reraised by ``join``.
+        will be reraised by :meth:`join`.
 
 
-        If ``timeout`` is not ``None`` and the operation takes
-        longer than ``timeout`` seconds, it will raise
-        the :class:`celery.timer.TimeoutError` exception.
+        :returns: list of return values for all tasks in the taskset.
 
 
         """
         """
         timeout_timer = TimeoutTimer(timeout) # Timeout timer starts here.
         timeout_timer = TimeoutTimer(timeout) # Timeout timer starts here.
@@ -292,7 +400,7 @@ class TaskSet(object):
         """Distribute processing of the arguments and collect the results
         """Distribute processing of the arguments and collect the results
         asynchronously.
         asynchronously.
         
         
-        Returns :class:`celery.result.AsyncResult` instance.
+        :returns: :class:`celery.result.AsyncResult` instance.
         
         
         """
         """
         serfunc = pickle.dumps(func)
         serfunc = pickle.dumps(func)
@@ -302,6 +410,8 @@ class TaskSet(object):
 def dmap(func, args, timeout=None):
 def dmap(func, args, timeout=None):
     """Distribute processing of the arguments and collect the results.
     """Distribute processing of the arguments and collect the results.
 
 
+    Example
+
         >>> from celery.task import map
         >>> from celery.task import map
         >>> import operator
         >>> import operator
         >>> dmap(operator.add, [[2, 2], [4, 4], [8, 8]])
         >>> dmap(operator.add, [[2, 2], [4, 4], [8, 8]])
@@ -312,7 +422,8 @@ def dmap(func, args, timeout=None):
 
 
 
 
 class AsynchronousMapTask(Task):
 class AsynchronousMapTask(Task):
-    """Task used internally by ``dmap_async``."""
+    """Task used internally by :func:`dmap_async` and
+    :meth:`TaskSet.map_async`.  """
     name = "celery.map_async"
     name = "celery.map_async"
 
 
     def run(self, serfunc, args, **kwargs):
     def run(self, serfunc, args, **kwargs):
@@ -325,7 +436,9 @@ def dmap_async(func, args, timeout=None):
     """Distribute processing of the arguments and collect the results
     """Distribute processing of the arguments and collect the results
     asynchronously.
     asynchronously.
     
     
-    Returns a :class:`celery.result.AsyncResult` object.
+    :returns: :class:`celery.result.AsyncResult` object.
+
+    Example
 
 
         >>> from celery.task import dmap_async
         >>> from celery.task import dmap_async
         >>> import operator
         >>> import operator
@@ -342,14 +455,21 @@ def dmap_async(func, args, timeout=None):
 
 
 
 
 class PeriodicTask(Task):
 class PeriodicTask(Task):
-    """A periodic task is a task that behaves like a cron job.
+    """A periodic task is a task that behaves like a :manpage:`cron` job.
 
 
-    The ``run_every`` attribute defines how often the task is run (its
-    interval), it can be either a ``datetime.timedelta`` object or a integer
-    specifying the time in seconds.
+    .. attribute:: run_every
+    
+        *REQUIRED* Defines how often the task is run (its interval),
+        it can be either a :class:`datetime.timedelta` object or an
+        integer specifying the time in seconds.
+
+    :raises NotImplementedError: if the :attr:`run_every` attribute is
+        not defined.
 
 
     You have to register the periodic task in the task registry.
     You have to register the periodic task in the task registry.
 
 
+    Example
+
         >>> from celery.task import tasks, PeriodicTask
         >>> from celery.task import tasks, PeriodicTask
         >>> from datetime import timedelta
         >>> from datetime import timedelta
         >>> class MyPeriodicTask(PeriodicTask):
         >>> class MyPeriodicTask(PeriodicTask):
@@ -378,17 +498,26 @@ class PeriodicTask(Task):
 
 
 
 
 class ExecuteRemoteTask(Task):
 class ExecuteRemoteTask(Task):
-    """Execute arbitrary function/object.
+    """Execute an arbitrary function or object.
+
+    *Note* You probably want :func:`execute_remote` instead, which this
+    is an internal component of.
 
 
     The object must be pickleable, so you can't use lambdas or functions
     The object must be pickleable, so you can't use lambdas or functions
-    defined in the REPL.
+    defined in the REPL (that is the python shell, or ``ipython``).
     
     
     """
     """
     name = "celery.execute_remote"
     name = "celery.execute_remote"
 
 
     def run(self, ser_callable, fargs, fkwargs, **kwargs):
     def run(self, ser_callable, fargs, fkwargs, **kwargs):
-        """Execute the pickled ``ser_callable``, with ``fargs`` as positional
-        arguments and ``fkwargs`` as keyword arguments."""
+        """
+        :param ser_callable: A pickled function or callable object. 
+
+        :param fargs: Positional arguments to apply to the function.
+
+        :param fkwargs: Keyword arguments to apply to the function.
+
+        """
         callable_ = pickle.loads(ser_callable)
         callable_ = pickle.loads(ser_callable)
         return callable_(*fargs, **fkwargs)
         return callable_(*fargs, **fkwargs)
 tasks.register(ExecuteRemoteTask)
 tasks.register(ExecuteRemoteTask)
@@ -396,11 +525,17 @@ tasks.register(ExecuteRemoteTask)
 
 
 def execute_remote(func, *args, **kwargs):
 def execute_remote(func, *args, **kwargs):
     """Execute arbitrary function/object remotely.
     """Execute arbitrary function/object remotely.
+        
+    :param func: A callable function or object.
+
+    :param \*args: Positional arguments to apply to the function.
+
+    :param \*\*kwargs: Keyword arguments to apply to the function.
 
 
     The object must be picklable, so you can't use lambdas or functions
     The object must be picklable, so you can't use lambdas or functions
     defined in the REPL (the objects must have an associated module).
     defined in the REPL (the objects must have an associated module).
 
 
-    :rtype: :class:`celery.result.AsyncResult`
+    :returns: class:`celery.result.AsyncResult`.
     
     
     """
     """
     return ExecuteRemoteTask.delay(pickle.dumps(func), args, kwargs)
     return ExecuteRemoteTask.delay(pickle.dumps(func), args, kwargs)
@@ -409,7 +544,9 @@ def execute_remote(func, *args, **kwargs):
 class DeleteExpiredTaskMetaTask(PeriodicTask):
 class DeleteExpiredTaskMetaTask(PeriodicTask):
     """A periodic task that deletes expired task metadata every day.
     """A periodic task that deletes expired task metadata every day.
 
 
-    This runs the current backend's cleanup() method.
+    This runs the current backend's
+    :meth:`celery.backends.base.BaseBackend.cleanup` method.
+
     """
     """
     name = "celery.delete_expired_task_meta"
     name = "celery.delete_expired_task_meta"
     run_every = timedelta(days=1)
     run_every = timedelta(days=1)

+ 8 - 5
celery/timer.py

@@ -1,4 +1,8 @@
-"""celery.timer"""
+"""
+
+Managing time and events
+
+"""
 import time
 import time
 
 
 
 
@@ -44,7 +48,7 @@ class EventTimer(object):
 
 
 
 
 class TimeoutTimer(object):
 class TimeoutTimer(object):
-    """A timer that raises :class:`TimeoutError` exception when the
+    """A timer that raises :exc:`TimeoutError` exception when the
     time has run out.
     time has run out.
    
    
     .. attribute:: timeout
     .. attribute:: timeout
@@ -64,9 +68,8 @@ class TimeoutTimer(object):
     def tick(self):
     def tick(self):
         """Run a timeout timer clock tick.
         """Run a timeout timer clock tick.
 
 
-        When ``timeout`` seconds has passed, it will raise a
-        :class:`TimeoutTimer` exception.
-        If ``timeout`` is not set, it will never time out.
+        :raises TimeoutError: when :attr:`timeout` seconds has passed.
+            If :attr:`timeout` is not set, it will never time out.
 
 
         """
         """
         if not self.timeout:
         if not self.timeout:

+ 0 - 13
docs/reference/celery.backends.rst

@@ -6,16 +6,3 @@ Backends - celery.backends
 
 
 .. automodule:: celery.backends
 .. automodule:: celery.backends
     :members:
     :members:
-
-.. function:: get_default_backend_cls()
-
-    Get the backend class specified in ``settings.CELERY_BACKEND``.
-
-.. class:: DefaultBackend
-
-    The backend class specified in ``settings.CELERY_BACKEND``.
-
-.. data:: default_backend
-
-    An instance of :class:`DefaultBackend`.
-    

+ 0 - 3
docs/reference/celery.registry.rst

@@ -6,6 +6,3 @@ Task Registry - celery.registry
 
 
 .. automodule:: celery.registry
 .. automodule:: celery.registry
     :members:
     :members:
-
-.. data:: tasks
-    The global task registry.