celery/docs/_ext/celerydocs.py

from docutils import nodes

from sphinx.environment import NoUri

APPATTRS = {
    'amqp': 'celery.app.amqp.AMQP',
    'backend': 'celery.backends.base.BaseBackend',
    'control': 'celery.app.control.Control',
    'events': 'celery.events.Events',
    'loader': 'celery.app.loaders.base.BaseLoader',
    'log': 'celery.app.log.Logging',
    'pool': 'kombu.connection.ConnectionPool',
    'tasks': 'celery.app.registry.Registry',

    'AsyncResult': 'celery.result.AsyncResult',
    'GroupResult': 'celery.result.GroupResult',
    'Worker': 'celery.apps.worker.Worker',
    'WorkController': 'celery.worker.WorkController',
    'Beat': 'celery.apps.beat.Beat',
    'Task': 'celery.app.task.Task',
    'send_task': 'celery.Celery.send_task',
    'connection': 'celery.Celery.connection',
}

ABBRS = {
    'Celery': 'celery.Celery',
}

ABBR_EMPTY = {
    'exc': 'celery.exceptions',
}
DEFAULT_EMPTY = 'celery.Celery'


def typeify(S, type):
    if type in ('meth', 'func'):
        return S + '()'
    return S


def shorten(S, newtarget, src_dict):
    if S.startswith('@-'):
        return S[2:]
    elif S.startswith('@'):
        if src_dict is APPATTRS:
            return '.'.join([pkg_of(newtarget), S[1:]])
        return S[1:]
    return S


def get_abbr(pre, rest, type):
    if pre:
        for d in APPATTRS, ABBRS:
            try:
                return d[pre], rest, d
            except KeyError:
                pass
        raise KeyError(pre)
    else:
        for d in APPATTRS, ABBRS:
            try:
                return d[rest], '', d
            except KeyError:
                pass
    return ABBR_EMPTY.get(type, DEFAULT_EMPTY), rest, ABBR_EMPTY


def resolve(S, type):
    if S.startswith('@'):
        S = S.lstrip('@-')
        try:
            pre, rest = S.split('.', 1)
        except ValueError:
            pre, rest = '', S

        target, rest, src = get_abbr(pre, rest, type)
        return '.'.join([target, rest]) if rest else target, src
    return S, None


def pkg_of(module_fqdn):
    return module_fqdn.split('.', 1)[0]


def basename(module_fqdn):
    return module_fqdn.lstrip('@').rsplit('.', -1)[-1]


def modify_textnode(T, newtarget, node, src_dict, type):
    src = node.children[0].rawsource
    return nodes.Text(
        (typeify(basename(T), type) if '~' in src
         else typeify(shorten(T, newtarget, src_dict), type)),
        src,
    )


def maybe_resolve_abbreviations(app, env, node, contnode):
    domainname = node.get('refdomain')
    target = node['reftarget']
    type = node['reftype']
    if target.startswith('@'):
        newtarget, src_dict = resolve(target, type)
        node['reftarget'] = newtarget
        # shorten text if '~' is not enabled.
        if len(contnode) and isinstance(contnode[0], nodes.Text):
                contnode[0] = modify_textnode(target, newtarget, node,
                                              src_dict, type)
        if domainname:
            try:
                domain = env.domains[node.get('refdomain')]
            except KeyError:
                raise NoUri
            return domain.resolve_xref(env, node['refdoc'], app.builder,
                                       type, newtarget,
                                       node, contnode)


def setup(app):
    app.connect('missing-reference', maybe_resolve_abbreviations)

    app.add_crossref_type(
        directivename='setting',
        rolename='setting',
        indextemplate='pair: %s; setting',
    )
    app.add_crossref_type(
        directivename='sig',
        rolename='sig',
        indextemplate='pair: %s; sig',
    )
    app.add_crossref_type(
        directivename='state',
        rolename='state',
        indextemplate='pair: %s; state',
    )
    app.add_crossref_type(
        directivename='control',
        rolename='control',
        indextemplate='pair: %s; control',
    )
    app.add_crossref_type(
        directivename='signal',
        rolename='signal',
        indextemplate='pair: %s; signal',
    )
    app.add_crossref_type(
        directivename='event',
        rolename='event',
        indextemplate='pair: %s; event',
    )