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

Source code for celery.contrib.sphinx

"""Sphinx documentation plugin used to document tasks.

Introduction
============

Usage
-----

The Celery extension for Sphinx requires Sphinx 2.0 or later.

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

.. code-block:: python

    extensions = (...,
                  'celery.contrib.sphinx')

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`
syntax.

Use ``.. autotask::`` to alternatively manually document a task.
"""
from inspect import signature

from sphinx.domains.python import PyFunction
from sphinx.ext.autodoc import FunctionDocumenter

from celery.app.task 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: sig = signature(wrapped) if "self" in sig.parameters or "cls" in sig.parameters: sig = sig.replace(parameters=list(sig.parameters.values())[1:]) return str(sig) 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 }