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

celery.utils

Utility functions.

Don’t import from here directly anymore, as these are only here for backwards compatibility.

class celery.utils.cached_property(fget=None, fset=None, fdel=None)

Implementation of Cached property.

deleter(fdel)

setter(fset)

celery.utils.chunks(it, n)

Split an iterator into chunks with n elements each.

Warning

it must be an actual iterator, if you pass this a concrete sequence will get you repeating elements.

So chunks(iter(range(1000)), 10) is fine, but chunks(range(1000), 10) is not.

Example

# n == 2 >>> x = chunks(iter([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]), 2) >>> list(x) [[0, 1], [2, 3], [4, 5], [6, 7], [8, 9], [10]]

# n == 3 >>> x = chunks(iter([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]), 3) >>> list(x) [[0, 1, 2], [3, 4, 5], [6, 7, 8], [9, 10]]

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

Generate task name from name/module pair.

celery.utils.gen_unique_id(_uuid: ~typing.Callable[[], ~uuid.UUID] = <function uuid4>) → str

Generate unique id in UUID4 format.

celery.utils.get_cls_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

celery.utils.get_full_cls_name(obj)

Return object name.

celery.utils.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.instantiate(name, *args, **kwargs)

Instantiate class by name.

See also

symbol_by_name().

celery.utils.memoize(maxsize=None, keyfun=None, Cache=<class 'kombu.utils.functional.LRUCache'>)

Decorator to cache function return value.

celery.utils.nodename(name: str, hostname: str) → str

Create node name from name/hostname pair.

celery.utils.nodesplit(name: str) → tuple[None, str] | list[str]

Split node name into tuple of name/hostname.

celery.utils.noop(*args, **kwargs)

No operation.

Takes any arguments/keyword arguments and does nothing.

celery.utils.uuid(_uuid: ~typing.Callable[[], ~uuid.UUID] = <function uuid4>) → str

Generate unique id in UUID4 format.

celery.utils.worker_direct(hostname: str | Queue) → Queue

Return the kombu.Queue being a direct route to a worker.

Parameters:

hostname (str, Queue) – The fully qualified node name of a worker (e.g., w1@example.com). If passed a kombu.Queue instance it will simply return that instead.