This document describes the current stable version of Celery (5.2). For development docs, go here.
celery.utils.imports
¶
Utilities related to importing modules and symbols by name.
-
exception
celery.utils.imports.
NotAPackage
[source]¶ Raised when importing a package, but it’s not a package.
-
celery.utils.imports.
cwd_in_path
()[source]¶ Context adding the current working directory to sys.path.
-
celery.utils.imports.
find_module
(module, path=None, imp=None)[source]¶ Version of
imp.find_module()
supporting dots.
-
celery.utils.imports.
gen_task_name
(app, name, module_name)[source]¶ Generate task name from name/module pair.
-
celery.utils.imports.
import_from_cwd
(module, imp=None, package=None)[source]¶ 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)[source]¶ Instantiate class by name.
See also
-
celery.utils.imports.
module_file
(module)[source]¶ Return the correct original file name of a module.
-
celery.utils.imports.
reload_from_cwd
(module, reloader=None)[source]¶ 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)[source]¶ 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