Function shell

celery/bin/shell.py:120–173

Start shell session with convenient access to celery symbols. The following symbols will be added to the main globals: - ``celery``: the current application. - ``chord``, ``group``, ``chain``, ``chunks``, ``xmap``, ``xstarmap`` ``subtask``, ``Task`` - all registered tasks.

(ctx, ipython=False, bpython=False,
          python=False, without_tasks=False, eventlet=False,
          gevent=False, **kwargs)

@click.pass_context
@handle_preload_options
def shell(ctx, ipython=False, bpython=False,
          python=False, without_tasks=False, eventlet=False,
          gevent=False, **kwargs):
    """Start shell session with convenient access to celery symbols.

    The following symbols will be added to the main globals:
    - ``celery``:  the current application.
    - ``chord``, ``group``, ``chain``, ``chunks``,
      ``xmap``, ``xstarmap`` ``subtask``, ``Task``
    - all registered tasks.
    """
    sys.path.insert(0, os.getcwd())
    if eventlet:
        import_module('celery.concurrency.eventlet')
    if gevent:
        import_module('celery.concurrency.gevent')
    import celery
    app = ctx.obj.app
    app.loader.import_default_modules()

    # pylint: disable=attribute-defined-outside-init
    locals = {
        'app': app,
        'celery': app,
        'Task': celery.Task,
        'chord': celery.chord,
        'group': celery.group,
        'chain': celery.chain,
        'chunks': celery.chunks,
        'xmap': celery.xmap,
        'xstarmap': celery.xstarmap,
        'subtask': celery.subtask,
        'signature': celery.signature,
    }

    if not without_tasks:
        locals.update({
            task.__name__: task for task in app.tasks.values()
            if not task.name.startswith('celery.')
        })

    if python:
        _invoke_fallback_shell(locals)
    elif bpython:
        try:
            _invoke_bpython_shell(locals)
        except ImportError:
            ctx.obj.echo(f'{ctx.obj.ERROR}: bpython is not installed')
    elif ipython:
        try:
            _invoke_ipython_shell(locals)
        except ImportError as e:
            ctx.obj.echo(f'{ctx.obj.ERROR}: {e}')
    _invoke_default_shell(locals)