This document describes the current stable version of Celery (5.7). For development docs, go here.
Source code for celery.backends.cache
"""Memcached and in-memory cache result backend."""
from kombu.utils.encoding import bytes_to_str, ensure_bytes
from kombu.utils.objects import cached_property
from celery.exceptions import ImproperlyConfigured
from celery.utils import deprecated
from celery.utils.functional import LRUCache
from celery.utils.log import get_logger
from .base import KeyValueStoreBackend
__all__ = ('CacheBackend',)
logger = get_logger(__name__)
# Memcached reads a longer expiry as an absolute Unix time.
MAX_RELATIVE_EXPIRES = 30 * 24 * 60 * 60
W_CHORD_EXPIRES_TOO_LONG = """\
result_chord_expires is over 30 days, which memcached reads as an absolute \
time, so chord keys will expire at once.\
"""
REQUIRES_BACKEND = """\
The Memcached backend requires the 'pymemcache' library. \
The 'pylibmc' and 'python-memcached' libraries are no longer used.\
"""
UNKNOWN_BACKEND = """\
The cache backend {0!r} is unknown,
Please use one of the following backends instead: {1}\
"""
# Options only supported by pylibmc, dropped before creating the client.
PYLIBMC_ONLY_OPTIONS = ('behaviors', 'binary')
# Defaults applied to the pymemcache client unless set in
# cache_backend_options: wait for the server reply so that failed
# writes raise, and don't block forever on an unresponsive server.
DEFAULT_CLIENT_OPTIONS = {
'default_noreply': False,
'connect_timeout': 5.0,
'timeout': 5.0,
}
# Global shared in-memory cache for in-memory cache client
# This is to share cache between threads
_DUMMY_CLIENT_CACHE = LRUCache(limit=5000)
def get_memcache_client():
"""Get pymemcache client factory."""
try:
from pymemcache.client.base import Client
from pymemcache.client.hash import HashClient
from pymemcache.client.retrying import RetryingClient
except ImportError:
raise ImproperlyConfigured(REQUIRES_BACKEND)
def ClientFactory(servers=None, **kwargs):
"""Create a pymemcache client with optional retry support.
Args:
servers: List of server addresses
**kwargs: Additional options including:
- retry_attempts: Number of retry attempts (enables RetryingClient)
- retry_delay: Delay between retries in seconds
- retry_for: List of exceptions to retry for
- do_not_retry_for: List of exceptions to not retry for
- behaviors, binary: Ignored for backward compatibility
with pylibmc
"""
# Remove pylibmc-specific options for backward compatibility
for option in PYLIBMC_ONLY_OPTIONS:
if option in kwargs:
kwargs.pop(option)
deprecated.warn(
description=f'The {option!r} cache backend option',
removal='6.0',
alternative='It is only supported by pylibmc and is ignored.',
stacklevel=3,
)
kwargs = {**DEFAULT_CLIENT_OPTIONS, **kwargs}
# Extract retry-related options
retry_attempts = kwargs.pop('retry_attempts', None)
retry_delay = kwargs.pop('retry_delay', None)
retry_for = kwargs.pop('retry_for', None)
do_not_retry_for = kwargs.pop('do_not_retry_for', None)
# Normalize/validate servers (CacheBackend passes a list, but guard other callers)
if servers is None:
servers = []
elif isinstance(servers, str):
servers = [servers]
elif (
isinstance(servers, tuple)
and len(servers) == 2
and isinstance(servers[0], str)
and isinstance(servers[1], int)
):
# Allow passing a single server address as (host, port)
servers = [servers]
elif not isinstance(servers, (list, tuple)):
raise ImproperlyConfigured(
f"Memcache servers must be a sequence of server addresses, got {type(servers)}: {servers!r}"
)
# Filter out empty entries (e.g. memcache:// or trailing ';')
servers = [s for s in servers if s]
if not servers:
raise ImproperlyConfigured("No memcache servers provided")
# Use HashClient for multiple servers, base Client for single server
if len(servers) > 1:
base_client = HashClient(servers, **kwargs)
else:
server = servers[0]
if isinstance(server, (str, tuple)):
base_client = Client(server, **kwargs)
else:
raise ImproperlyConfigured(
f"Memcache server address must be a string or tuple, got {type(server)}: {server!r}"
)
# Wrap with RetryingClient if retry options are specified
if retry_attempts is not None:
retry_kwargs = {'attempts': retry_attempts}
if retry_delay is not None:
retry_kwargs['retry_delay'] = retry_delay
if retry_for is not None:
retry_kwargs['retry_for'] = retry_for
if do_not_retry_for is not None:
retry_kwargs['do_not_retry_for'] = do_not_retry_for
return RetryingClient(base_client, **retry_kwargs)
return base_client
return ClientFactory, bytes_to_str
def import_best_memcache():
"""Deprecated: use :func:`get_memcache_client` instead."""
deprecated.warn(description='import_best_memcache()', removal='6.0',
alternative='Use get_memcache_client() instead.')
try:
import pymemcache
except ImportError:
raise ImproperlyConfigured(REQUIRES_BACKEND)
return False, pymemcache, bytes_to_str
def get_best_memcache(*args, **kwargs):
"""Deprecated: use :func:`get_memcache_client` instead."""
deprecated.warn(description='get_best_memcache()', removal='6.0',
alternative='Use get_memcache_client() instead.')
return get_memcache_client()
class DummyClient:
def __init__(self, *args, **kwargs):
self.cache = _DUMMY_CLIENT_CACHE
def get(self, key, *args, **kwargs):
return self.cache.get(key)
def get_multi(self, keys):
cache = self.cache
return {k: cache[k] for k in keys if k in cache}
def set(self, key, value, *args, **kwargs):
self.cache[key] = value
def delete(self, key, *args, **kwargs):
self.cache.pop(key, None)
def incr(self, key, delta=1):
return self.cache.incr(key, delta)
def touch(self, key, expire):
pass
backends = {
'memcache': get_memcache_client,
'memcached': get_memcache_client,
'pylibmc': get_memcache_client, # Backward compatibility
'pymemcache': get_memcache_client,
'memory': lambda: (DummyClient, ensure_bytes),
}
[docs]
class CacheBackend(KeyValueStoreBackend):
"""Cache result backend."""
servers = None
supports_autoexpire = True
supports_native_join = True
implements_incr = True
def __init__(self, app, expires=None, backend=None,
options=None, url=None, **kwargs):
options = {} if not options else options
super().__init__(app, **kwargs)
self.url = url
self.options = dict(self.app.conf.cache_backend_options,
**options)
self.backend = url or backend or self.app.conf.cache_backend
if self.backend:
self.backend, _, servers = self.backend.partition('://')
self.servers = servers.rstrip('/').split(';')
self.expires = self.prepare_expires(expires, type=int)
try:
self.Client, self.key_t = backends[self.backend]()
except KeyError:
raise ImproperlyConfigured(UNKNOWN_BACKEND.format(
self.backend, ', '.join(backends)))
self._encode_prefixes() # rencode the keyprefixes
if (self._chord_expires is not None
and self.chord_expires > MAX_RELATIVE_EXPIRES):
logger.warning(W_CHORD_EXPIRES_TOO_LONG)
def _apply_chord_incr(self, header_result_args, body, **kwargs):
chord_key = self.get_key_for_chord(header_result_args[0])
self.client.set(chord_key, 0, self.chord_expires or 0)
return super()._apply_chord_incr(
header_result_args, body, **kwargs)
@cached_property
def client(self):
return self.Client(self.servers, **self.options)
def __reduce__(self, args=(), kwargs=None):
kwargs = {} if not kwargs else kwargs
servers = ';'.join(self.servers)
backend = f'{self.backend}://{servers}/'
kwargs.update(
{'backend': backend,
'expires': self.expires,
'options': self.options})
return super().__reduce__(args, kwargs)
[docs]
def as_uri(self, *args, **kwargs):
"""Return the backend as an URI.
This properly handles the case of multiple servers.
"""
servers = ';'.join(self.servers)
return f'{self.backend}://{servers}/'
