This document describes the current stable version of Celery (5.6). 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.