This document describes the current stable version of Celery (4.4). For development docs, go here.

Source code for celery.contrib.sphinx

# -*- coding: utf-8 -*-
"""Sphinx documentation plugin used to document tasks.



Add the extension to your :file:`docs/` configuration module:

.. code-block:: python

    extensions = (...,

If you'd like to change the prefix for tasks in reference documentation
then you can change the ``celery_task_prefix`` configuration value:

.. code-block:: python

    celery_task_prefix = '(task)'  # < default

With the extension installed `autodoc` will automatically find
task decorated objects (e.g. when using the automodule directive)
and generate the correct (as well as add a ``(task)`` prefix),
and you can also refer to the tasks using `:task:proj.tasks.add`

Use ``.. autotask::`` to alternatively manually document a task.
from __future__ import absolute_import, unicode_literals

from import PyFunction
from sphinx.ext.autodoc import FunctionDocumenter

from import BaseTask

try:  # pragma: no cover
    from inspect import formatargspec, getfullargspec
except ImportError:  # Py2
    from inspect import formatargspec, getargspec as getfullargspec  # noqa

[docs]class TaskDocumenter(FunctionDocumenter): """Document task definitions.""" objtype = 'task' member_order = 11
[docs] @classmethod def can_document_member(cls, member, membername, isattr, parent): return isinstance(member, BaseTask) and getattr(member, '__wrapped__')
[docs] def format_args(self): wrapped = getattr(self.object, '__wrapped__', None) if wrapped is not None: argspec = getfullargspec(wrapped) if argspec[0] and argspec[0][0] in ('cls', 'self'): del argspec[0][0] fmt = formatargspec(*argspec) fmt = fmt.replace('\\', '\\\\') return fmt return ''
[docs] def document_members(self, all_members=False): pass
[docs] def check_module(self): # Normally checks if *self.object* is really defined in the module # given by *self.modname*. But since functions decorated with the @task # decorator are instances living in the celery.local, we have to check # the wrapped function instead. wrapped = getattr(self.object, '__wrapped__', None) if wrapped and getattr(wrapped, '__module__') == self.modname: return True return super(TaskDocumenter, self).check_module()
[docs]class TaskDirective(PyFunction): """Sphinx task directive."""
[docs] def get_signature_prefix(self, sig): return self.env.config.celery_task_prefix
[docs]def autodoc_skip_member_handler(app, what, name, obj, skip, options): """Handler for autodoc-skip-member event.""" # Celery tasks created with the @task decorator have the property # that *obj.__doc__* and *obj.__class__.__doc__* are equal, which # trips up the logic in sphinx.ext.autodoc that is supposed to # suppress repetition of class documentation in an instance of the # class. This overrides that behavior. if isinstance(obj, BaseTask) and getattr(obj, '__wrapped__'): if skip: return False return None
[docs]def setup(app): """Setup Sphinx extension.""" app.setup_extension('sphinx.ext.autodoc') app.add_autodocumenter(TaskDocumenter) app.add_directive_to_domain('py', 'task', TaskDirective) app.add_config_value('celery_task_prefix', '(task)', True) app.connect('autodoc-skip-member', autodoc_skip_member_handler) return { 'parallel_read_safe': True }