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

# `celery.utils.imports`

Utilities related to importing modules and symbols by name.

exception celery.utils.imports.NotAPackage

  Raised when importing a package, but it’s not a package.

celery.utils.imports.cwd_in_path()

  Context adding the current working directory to sys.path.

celery.utils.imports.find_module(module, path=None, imp=None)

  Version of `imp.find_module()` supporting dots.

celery.utils.imports.gen_task_name(app, name, module_name)

  Generate task name from name/module pair.

celery.utils.imports.import_from_cwd(module, imp=None, package=None)

  Import module, temporarily including modules in the current directory.

  Modules located in the current directory has precedence over modules located in sys.path.

celery.utils.imports.instantiate(name, *args, **kwargs)

  Instantiate class by name.

  See also

  `symbol_by_name()`.

celery.utils.imports.module_file(module)

  Return the correct original file name of a module.

celery.utils.imports.qualname(obj)

  Return object name.

celery.utils.imports.reload_from_cwd(module, reloader=None)

  Reload module (ensuring that CWD is in sys.path).

celery.utils.imports.symbol_by_name(name, aliases=None, imp=None, package=None, sep='.', default=None, **kwargs)

  Get symbol by qualified name.

  The name should be the full dot-separated path to the class:

      modulename.ClassName

  Example:

      celery.concurrency.processes.TaskPool
                                  ^- class name

  or using ‘:’ to separate module and symbol:

      celery.concurrency.processes:TaskPool

  If aliases is provided, a dict containing short name/long name mappings, the name is looked up in the aliases first.

  Examples

      >>> symbol_by_name('celery.concurrency.processes.TaskPool')
      <class 'celery.concurrency.processes.TaskPool'>

      >>> symbol_by_name('default', {
      ...     'default': 'celery.concurrency.processes.TaskPool'})
      <class 'celery.concurrency.processes.TaskPool'>

  # Does not try to look up non-string names. >>> from celery.concurrency.processes import TaskPool >>> symbol_by_name(TaskPool) is TaskPool True