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

Source code for celery.contrib.sphinx

"""Sphinx documentation plugin used to document tasks.



The Celery extension for Sphinx requires Sphinx 2.0 or later.

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 inspect import formatargspec, getfullargspec

from import PyFunction
from sphinx.ext.autodoc import FunctionDocumenter

from import BaseTask

[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().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 }