This document describes the current stable version of Celery (5.7). For development docs, go here.
Source code for celery.backends.database
"""SQLAlchemy result store backend."""
import logging
import pickle
from contextlib import contextmanager
from kombu.utils.encoding import ensure_bytes
from celery import states
from celery.backends.base import BaseBackend
from celery.exceptions import ImproperlyConfigured
from celery.result import result_from_tuple
from celery.utils.imports import symbol_by_name
from celery.utils.time import maybe_timedelta
from .models import Task, TaskExtended, TaskSet
from .session import SessionManager
try:
from sqlalchemy.exc import DatabaseError, InterfaceError, InvalidRequestError
from sqlalchemy.orm import defer
from sqlalchemy.orm.exc import StaleDataError
except ImportError:
raise ImproperlyConfigured(
'The database result backend requires SQLAlchemy to be installed. '
'See https://pypi.org/project/SQLAlchemy/')
logger = logging.getLogger(__name__)
__all__ = ('DatabaseBackend',)
RETRYABLE_DB_ERRORS = (
DatabaseError,
InterfaceError,
InvalidRequestError,
StaleDataError,
)
@contextmanager
def session_cleanup(session):
try:
yield
except Exception:
session.rollback()
raise
finally:
session.close()
[docs]
class DatabaseBackend(BaseBackend):
"""The database result backend."""
# ResultSet.iterate should sleep this much between each pool,
# to not bombard the database with queries.
subpolling_interval = 0.5
task_cls = Task
taskset_cls = TaskSet
def __init__(self, dburi=None, engine_options=None, url=None, **kwargs):
# The `url` argument was added later and is used by
# the app to set backend by url (celery.app.backends.by_url)
super().__init__(expires_type=maybe_timedelta,
url=url, **kwargs)
conf = self.app.conf
# Override retry defaults to preserve backward compatibility.
# Previously, DatabaseBackend used a custom @retry decorator that always
# retried with max_retries=3. We maintain this behavior by default.
self.always_retry = conf.get('result_backend_always_retry', True)
self.max_retries = conf.get('result_backend_max_retries', 3)
if self.extended_result:
self.task_cls = TaskExtended
self.url = url or dburi or conf.database_url
# Merge engine options: defaults from config <- constructor overrides
# The defaults (pool_pre_ping=True, pool_recycle=3600) are defined in
# celery/app/defaults.py under database_engine_options
self.engine_options = dict(
conf.database_engine_options or {},
**(engine_options or {})
)
self.short_lived_sessions = kwargs.get(
'short_lived_sessions',
conf.database_short_lived_sessions)
schemas = conf.database_table_schemas or {}
tablenames = conf.database_table_names or {}
self.task_cls.configure(
schema=schemas.get('task'),
name=tablenames.get('task'))
self.taskset_cls.configure(
schema=schemas.get('group'),
name=tablenames.get('group'))
if not self.url:
raise ImproperlyConfigured(
'Missing connection string! Do you have the'
' database_url setting set to a real value?')
engine_callback = kwargs.get(
'engine_callback', conf.database_engine_callback)
if isinstance(engine_callback, str):
engine_callback = symbol_by_name(engine_callback)
if engine_callback is not None and not callable(engine_callback):
raise ImproperlyConfigured(
'database_engine_callback must be callable, got {!r}'.format(
engine_callback))
self.session_manager = SessionManager(
engine_callback=engine_callback)
create_tables_at_setup = conf.database_create_tables_at_setup
if create_tables_at_setup is True:
self._create_tables()
@property
def extended_result(self):
return self.app.conf.find_value_for_key('extended', 'result')
def _create_tables(self):
"""Create the task and taskset tables."""
self._ensure_retryable(self.ResultSession)
[docs]
def ResultSession(self, session_manager=None):
if session_manager is None:
session_manager = self.session_manager
return session_manager.session_factory(
dburi=self.url,
short_lived_sessions=self.short_lived_sessions,
**self.engine_options)
def _query_task(self, session, task_id):
"""Query task by id, falling back to deferring missing columns from database."""
try:
tasks = list(session.query(self.task_cls).filter(self.task_cls.task_id == task_id))
return tasks and tasks[0]
except DatabaseError as exc:
exc_str = str(exc).lower()
defers = []
if 'children' in exc_str and hasattr(self.task_cls, 'children'):
defers.append(defer(self.task_cls.children))
if 'stamps' in exc_str and hasattr(self.task_cls, 'stamps'):
defers.append(defer(self.task_cls.stamps))
if defers:
tasks = list(session.query(self.task_cls).options(*defers).filter(
self.task_cls.task_id == task_id
))
return tasks and tasks[0]
raise
def _store_result(self, task_id, result, state, traceback=None,
request=None, **kwargs):
"""Store return value and state of an executed task."""
session = self.ResultSession()
with session_cleanup(session):
task = self._query_task(session, task_id)
if not task:
task = self.task_cls(task_id)
task.task_id = task_id
session.add(task)
session.flush()
self._update_result(task, result, state, traceback=traceback, request=request)
session.commit()
def _update_result(self, task, result, state, traceback=None,
request=None):
meta = self._get_result_meta(result=ensure_bytes(self.encode(result)), state=state,
traceback=traceback, request=request,
format_date=False, encode=True)
# Exclude the primary key id, task_id, children, and stamps columns
# as we should not set it None or handle them separately
columns = [column.name for column in self.task_cls.__table__.columns
if column.name not in {'id', 'task_id', 'children', 'stamps'}]
# Iterate through the columns name of the table
# to set the value from meta.
# If the value is not present in meta, set None
for column in columns:
value = meta.get(column)
setattr(task, column, value)
if hasattr(task, 'children') and 'children' in self.task_cls.__table__.columns:
children = meta.get('children')
if children:
setattr(task, 'children', ensure_bytes(self.encode(children)))
else:
setattr(task, 'children', None)
if hasattr(task, 'stamps') and 'stamps' in self.task_cls.__table__.columns:
stamped_headers = meta.get('stamped_headers')
if stamped_headers:
stamps_data = {
h: meta.get(h) for h in stamped_headers if h in meta
}
stamps_info = {
'stamped_headers': stamped_headers,
'stamps': stamps_data,
}
setattr(task, 'stamps', ensure_bytes(self.encode(stamps_info)))
elif getattr(task, 'stamps', None) is None:
setattr(task, 'stamps', None)
def _get_task_meta_for(self, task_id):
"""Get task meta-data for a task by id."""
session = self.ResultSession()
with session_cleanup(session):
task = self._query_task(session, task_id)
if not task:
task = self.task_cls(task_id)
task.status = states.PENDING
task.result = None
data = task.to_dict()
data['result'] = self._decode_stored_result(data['result'])
if data.get('args', None) is not None:
data['args'] = self.decode(data['args'])
if data.get('kwargs', None) is not None:
data['kwargs'] = self.decode(data['kwargs'])
if data.get('children', None) is not None:
data['children'] = self.decode(data['children'])
raw_stamps = data.pop('stamps', None)
if raw_stamps is not None:
try:
stamps_info = self.decode(raw_stamps)
if isinstance(stamps_info, dict):
if 'stamped_headers' in stamps_info:
data['stamped_headers'] = stamps_info['stamped_headers']
if 'stamps' in stamps_info and isinstance(stamps_info['stamps'], dict):
data.update(stamps_info['stamps'])
except Exception:
pass
return self.meta_from_decoded(data)
def _decode_stored_result(self, payload):
"""Decode a value stored in the ``result`` column.
Rows written before the fix for celery/celery#3025 always stored a
raw pickle blob regardless of the configured ``result_serializer``,
so a row that can't be decoded with the current serializer, but
looks like a pickle payload (starts with the pickle protocol 2+
marker that every pickle Celery/kombu produces starts with), is
assumed to be one of those and is unpickled directly instead.
Anything else re-raises the original decode error instead of
blindly unpickling arbitrary bytes.
"""
if payload is None:
return payload
payload = ensure_bytes(payload)
try:
return self.decode(payload)
except Exception:
# Legacy rows written before celery/celery#3025 were stored via
# SQLAlchemy's PickleType, which uses a binary pickle protocol.
if payload[:1] != b'\x80':
raise
logger.warning(
'Task result payload could not be decoded using the '
'configured result_serializer; falling back to pickle '
'for backward compatibility with pre-fix data.',
exc_info=True,
)
return pickle.loads(payload)
[docs]
def task_result_exists(self, task_id):
"""Check if a result exists in the database for the given task ID.
.. versionadded:: 5.7.0
"""
return self._ensure_retryable(
self._task_result_exists, task_id=task_id)
def _task_result_exists(self, task_id):
session = self.ResultSession()
with session_cleanup(session):
return session.query(self.task_cls).filter(
self.task_cls.task_id == task_id
).first() is not None
def _save_group(self, group_id, result):
"""Store the result of an executed group."""
session = self.ResultSession()
with session_cleanup(session):
value = ensure_bytes(self.encode(self.prepare_value(result)))
group = self.taskset_cls(group_id, value)
session.add(group)
session.flush()
session.commit()
return result
def _restore_group(self, group_id):
"""Get meta-data for group by id."""
session = self.ResultSession()
with session_cleanup(session):
group = session.query(self.taskset_cls).filter(
self.taskset_cls.taskset_id == group_id).first()
if group:
data = group.to_dict()
value = self._decode_stored_result(data['result'])
data['result'] = (
value if value is None else result_from_tuple(value, self.app)
)
return data
def _delete_group(self, group_id):
"""Delete meta-data for group by id."""
session = self.ResultSession()
with session_cleanup(session):
session.query(self.taskset_cls).filter(
self.taskset_cls.taskset_id == group_id).delete()
session.flush()
session.commit()
def _forget(self, task_id):
"""Forget about result."""
session = self.ResultSession()
with session_cleanup(session):
session.query(self.task_cls).filter(self.task_cls.task_id == task_id).delete()
session.commit()
def _cleanup(self):
session = self.ResultSession()
expires = self.expires
now = self.app.now()
with session_cleanup(session):
session.query(self.task_cls).filter(
self.task_cls.date_done < (now - expires)).delete()
session.query(self.taskset_cls).filter(
self.taskset_cls.date_done < (now - expires)).delete()
session.commit()
def __reduce__(self, args=(), kwargs=None):
kwargs = {} if not kwargs else kwargs
kwargs.update(
{'dburi': self.url,
'expires': self.expires,
'engine_options': self.engine_options})
return super().__reduce__(args, kwargs)
