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

Source code for celery.backends.redis

# -*- coding: utf-8 -*-

    Redis result store backend.

from __future__ import absolute_import

from functools import partial

from kombu.utils import cached_property, retry_over_time
from kombu.utils.url import _parse_url

from celery import states
from celery.canvas import maybe_signature
from celery.exceptions import ChordError, ImproperlyConfigured
from celery.five import string_t
from celery.utils import deprecated_property, strtobool
from celery.utils.functional import dictfilter
from celery.utils.log import get_logger
from celery.utils.timeutils import humanize_seconds

from .base import KeyValueStoreBackend

    import redis
    from redis.exceptions import ConnectionError
    from kombu.transport.redis import get_redis_error_classes
except ImportError:                 # pragma: no cover
    redis = None                    # noqa
    ConnectionError = None          # noqa
    get_redis_error_classes = None  # noqa

__all__ = ['RedisBackend']

You need to install the redis library in order to use \
the Redis result store backend."""

logger = get_logger(__name__)
error = logger.error

[docs]class RedisBackend(KeyValueStoreBackend): """Redis task result store.""" #: redis-py client module. redis = redis #: Maximium number of connections in the pool. max_connections = None supports_autoexpire = True supports_native_join = True implements_incr = True def __init__(self, host=None, port=None, db=None, password=None, expires=None, max_connections=None, url=None, connection_pool=None, new_join=False, **kwargs): super(RedisBackend, self).__init__(**kwargs) conf = if self.redis is None: raise ImproperlyConfigured(REDIS_MISSING) self._client_capabilities = self._detect_client_capabilities() # For compatibility with the old REDIS_* configuration keys. def _get(key): for prefix in 'CELERY_REDIS_{0}', 'REDIS_{0}': try: return conf[prefix.format(key)] except KeyError: pass if host and '://' in host: url = host host = None self.max_connections = ( max_connections or _get('MAX_CONNECTIONS') or self.max_connections ) self._ConnectionPool = connection_pool self.connparams = { 'host': _get('HOST') or 'localhost', 'port': _get('PORT') or 6379, 'db': _get('DB') or 0, 'password': _get('PASSWORD'), 'max_connections': self.max_connections, } if url: self.connparams = self._params_from_url(url, self.connparams) self.url = url self.expires = self.prepare_expires(expires, type=int) try: new_join = strtobool(self.connparams.pop('new_join')) except KeyError: pass if new_join: self.apply_chord = self._new_chord_apply self.on_chord_part_return = self._new_chord_return self.connection_errors, self.channel_errors = ( get_redis_error_classes() if get_redis_error_classes else ((), ())) def _params_from_url(self, url, defaults): scheme, host, port, user, password, path, query = _parse_url(url) connparams = dict( defaults, **dictfilter({ 'host': host, 'port': port, 'password': password, 'db': query.pop('virtual_host', None)}) ) if scheme == 'socket': # use 'path' as path to the socket… in this case # the database number should be given in 'query' connparams.update({ 'connection_class': self.redis.UnixDomainSocketConnection, 'path': '/' + path, }) # host+port are invalid options when using this connection type. connparams.pop('host', None) connparams.pop('port', None) else: connparams['db'] = path # db may be string and start with / like in kombu. db = connparams.get('db') or 0 db = db.strip('/') if isinstance(db, string_t) else db connparams['db'] = int(db) # Query parameters override other parameters connparams.update(query) return connparams
[docs] def get(self, key): return self.client.get(key)
[docs] def mget(self, keys): return self.client.mget(keys)
[docs] def ensure(self, fun, args, **policy): retry_policy = dict(self.retry_policy, **policy) max_retries = retry_policy.get('max_retries') return retry_over_time( fun, self.connection_errors, args, {}, partial(self.on_connection_error, max_retries), **retry_policy )
[docs] def on_connection_error(self, max_retries, exc, intervals, retries): tts = next(intervals) error('Connection to Redis lost: Retry (%s/%s) %s.', retries, max_retries or 'Inf', humanize_seconds(tts, 'in ')) return tts
[docs] def set(self, key, value, **retry_policy): return self.ensure(self._set, (key, value), **retry_policy)
def _set(self, key, value): with self.client.pipeline() as pipe: if self.expires: pipe.setex(key, value, self.expires) else: pipe.set(key, value) pipe.publish(key, value) pipe.execute()
[docs] def delete(self, key): self.client.delete(key)
[docs] def incr(self, key): return self.client.incr(key)
[docs] def expire(self, key, value): return self.client.expire(key, value)
def _unpack_chord_result(self, tup, decode, EXCEPTION_STATES=states.EXCEPTION_STATES, PROPAGATE_STATES=states.PROPAGATE_STATES): _, tid, state, retval = decode(tup) if state in EXCEPTION_STATES: retval = self.exception_to_python(retval) if state in PROPAGATE_STATES: raise ChordError('Dependency {0} raised {1!r}'.format(tid, retval)) return retval def _new_chord_apply(self, header, partial_args, group_id, body, result=None, **options): # avoids saving the group in the redis db. return header(*partial_args, task_id=group_id) def _new_chord_return(self, task, state, result, propagate=None, PROPAGATE_STATES=states.PROPAGATE_STATES): app = if propagate is None: propagate = request = task.request tid, gid =, if not gid or not tid: return client = self.client jkey = self.get_key_for_group(gid, '.j') result = self.encode_result(result, state) with client.pipeline() as pipe: _, readycount, _ = pipe \ .rpush(jkey, self.encode([1, tid, state, result])) \ .llen(jkey) \ .expire(jkey, 86400) \ .execute() try: callback = maybe_signature(request.chord, app=app) total = callback['chord_size'] if readycount == total: decode, unpack = self.decode, self._unpack_chord_result with client.pipeline() as pipe: resl, _, = pipe \ .lrange(jkey, 0, total) \ .delete(jkey) \ .execute() try: callback.delay([unpack(tup, decode) for tup in resl]) except Exception as exc: error('Chord callback for %r raised: %r',, exc, exc_info=1) return self.chord_error_from_stack( callback, ChordError('Callback error: {0!r}'.format(exc)), ) except ChordError as exc: error('Chord %r raised: %r',, exc, exc_info=1) return self.chord_error_from_stack(callback, exc) except Exception as exc: error('Chord %r raised: %r',, exc, exc_info=1) return self.chord_error_from_stack( callback, ChordError('Join error: {0!r}'.format(exc)), ) def _detect_client_capabilities(self, socket_connect_timeout=False): if self.redis.VERSION < (2, 4, 4): raise ImproperlyConfigured( 'Redis backend requires redis-py versions 2.4.4 or later. ' 'You have {0.__version__}'.format(redis)) if self.redis.VERSION >= (2, 10): socket_connect_timeout = True return {'socket_connect_timeout': socket_connect_timeout} def _create_client(self, socket_timeout=None, socket_connect_timeout=None, **params): return self._new_redis_client( socket_timeout=socket_timeout and float(socket_timeout), socket_connect_timeout=socket_connect_timeout and float( socket_connect_timeout), **params ) def _new_redis_client(self, **params): if not self._client_capabilities['socket_connect_timeout']: params.pop('socket_connect_timeout', None) return self.redis.Redis(connection_pool=self.ConnectionPool(**params)) @property def ConnectionPool(self): if self._ConnectionPool is None: self._ConnectionPool = self.redis.ConnectionPool return self._ConnectionPool @cached_property
[docs] def client(self): return self._create_client(**self.connparams)
def __reduce__(self, args=(), kwargs={}): return super(RedisBackend, self).__reduce__( (self.url, ), {'expires': self.expires}, ) @deprecated_property(3.2, 3.3)
[docs] def host(self): return self.connparams['host']
@deprecated_property(3.2, 3.3)
[docs] def port(self): return self.connparams['port']
@deprecated_property(3.2, 3.3)
[docs] def db(self): return self.connparams['db']
@deprecated_property(3.2, 3.3)
[docs] def password(self): return self.connparams['password']