View
184 celery/app/base.py

Large diffs are not rendered by default.

Oops, something went wrong.
View
2 celery/app/builtins.py
@@ -54,7 +54,7 @@ def add_unlock_chord_task(app):
from celery.exceptions import ChordError
from celery.result import allow_join_result, result_from_tuple
default_propagate = app.conf.CELERY_CHORD_PROPAGATES
default_propagate = app.conf.chord_propagates
@app.task(name='celery.chord_unlock', max_retries=None, shared=False,
default_retry_delay=1, ignore_result=True, lazy=False, bind=True)
View
440 celery/app/defaults.py

Large diffs are not rendered by default.

Oops, something went wrong.
View
8 celery/app/log.py
@@ -59,9 +59,9 @@ class Logging(object):
def __init__(self, app):
self.app = app
self.loglevel = mlevel(logging.WARN)
self.format = self.app.conf.CELERYD_LOG_FORMAT
self.task_format = self.app.conf.CELERYD_TASK_LOG_FORMAT
self.colorize = self.app.conf.CELERYD_LOG_COLOR
self.format = self.app.conf.worker_log_format
self.task_format = self.app.conf.worker_task_log_format
self.colorize = self.app.conf.worker_log_color
def setup(self, loglevel=None, logfile=None, redirect_stdouts=False,
redirect_level='WARNING', colorize=None, hostname=None):
@@ -105,7 +105,7 @@ def setup_logging_subsystem(self, loglevel=None, logfile=None, format=None,
if not receivers:
root = logging.getLogger()
if self.app.conf.CELERYD_HIJACK_ROOT_LOGGER:
if self.app.conf.worker_hijack_root_logger:
root.handlers = []
get_logger('celery').handlers = []
get_logger('celery.task').handlers = []
View
8 celery/app/routes.py
@@ -4,7 +4,7 @@
~~~~~~~~~~~~~
Contains utilities for working with task routers,
(:setting:`CELERY_ROUTES`).
(:setting:`task_routes`).
"""
from __future__ import absolute_import
@@ -52,7 +52,7 @@ def route(self, options, task, args=(), kwargs={}):
return lpmerge(self.expand_destination(route), options)
if 'queue' not in options:
options = lpmerge(self.expand_destination(
self.app.conf.CELERY_DEFAULT_QUEUE), options)
self.app.conf.task_default_queue), options)
return options
def expand_destination(self, route):
@@ -72,15 +72,15 @@ def expand_destination(self, route):
route['queue'] = self.queues[queue]
except KeyError:
raise QueueNotFound(
'Queue {0!r} missing from CELERY_QUEUES'.format(queue))
'Queue {0!r} missing from task_queues'.format(queue))
return route
def lookup_route(self, task, args=None, kwargs=None):
return _first_route(self.routes, task, args, kwargs)
def prepare(routes):
"""Expands the :setting:`CELERY_ROUTES` setting."""
"""Expands the :setting:`task_routes` setting."""
def expand_route(route):
if isinstance(route, dict):
View
49 celery/app/task.py
@@ -160,7 +160,7 @@ class Task(object):
rate_limit = None
#: If enabled the worker will not store task state and return values
#: for this task. Defaults to the :setting:`CELERY_IGNORE_RESULT`
#: for this task. Defaults to the :setting:`task_ignore_result`
#: setting.
ignore_result = None
@@ -173,7 +173,7 @@ class Task(object):
#: configured to ignore results.
store_errors_even_if_ignored = None
#: If enabled an email will be sent to :setting:`ADMINS` whenever a task
#: If enabled an email will be sent to :setting:`admins` whenever a task
#: of this type fails.
send_error_emails = None
@@ -182,11 +182,11 @@ class Task(object):
serializer = None
#: Hard time limit.
#: Defaults to the :setting:`CELERYD_TASK_TIME_LIMIT` setting.
#: Defaults to the :setting:`task_time_limit` setting.
time_limit = None
#: Soft time limit.
#: Defaults to the :setting:`CELERYD_TASK_SOFT_TIME_LIMIT` setting.
#: Defaults to the :setting:`task_soft_time_limit` setting.
soft_time_limit = None
#: The result store backend used for this task.
@@ -205,7 +205,7 @@ class Task(object):
#: running.
#:
#: The application default can be overridden using the
#: :setting:`CELERY_TRACK_STARTED` setting.
#: :setting:`task_track_started` setting.
track_started = None
#: When enabled messages for this task will be acknowledged **after**
@@ -217,7 +217,7 @@ class Task(object):
#: applications).
#:
#: The application default can be overridden with the
#: :setting:`CELERY_ACKS_LATE` setting.
#: :setting:`task_acks_late` setting.
acks_late = None
#: Even if :attr:`acks_late` is enabled, the worker will
@@ -255,15 +255,14 @@ class Task(object):
__bound__ = False
from_config = (
('send_error_emails', 'CELERY_SEND_TASK_ERROR_EMAILS'),
('serializer', 'CELERY_TASK_SERIALIZER'),
('rate_limit', 'CELERY_DEFAULT_RATE_LIMIT'),
('track_started', 'CELERY_TRACK_STARTED'),
('acks_late', 'CELERY_ACKS_LATE'),
('reject_on_worker_lost', 'CELERY_REJECT_ON_WORKER_LOST'),
('ignore_result', 'CELERY_IGNORE_RESULT'),
('store_errors_even_if_ignored',
'CELERY_STORE_ERRORS_EVEN_IF_IGNORED'),
('send_error_emails', 'task_send_error_emails'),
('serializer', 'task_serializer'),
('rate_limit', 'task_default_rate_limit'),
('track_started', 'task_track_started'),
('acks_late', 'task_acks_late'),
('reject_on_worker_lost', 'task_reject_on_worker_lost'),
('ignore_result', 'task_ignore_result'),
('store_errors_even_if_ignored', 'task_store_errors_even_if_ignored'),
)
#: ignored
@@ -409,12 +408,12 @@ def apply_async(self, args=None, kwargs=None, task_id=None, producer=None,
:keyword retry: If enabled sending of the task message will be retried
in the event of connection loss or failure. Default
is taken from the :setting:`CELERY_TASK_PUBLISH_RETRY`
is taken from the :setting:`task_publish_retry`
setting. Note that you need to handle the
producer/connection manually for this to work.
:keyword retry_policy: Override the retry policy used. See the
:setting:`CELERY_TASK_PUBLISH_RETRY_POLICY`
:setting:`task_publish_retry_policy`
setting.
:keyword routing_key: Custom routing key used to route the task to a
@@ -423,8 +422,8 @@ def apply_async(self, args=None, kwargs=None, task_id=None, producer=None,
routing keys to topic exchanges.
:keyword queue: The queue to route the task to. This must be a key
present in :setting:`CELERY_QUEUES`, or
:setting:`CELERY_CREATE_MISSING_QUEUES` must be
present in :setting:`task_queues`, or
:setting:`task_create_missing_queues` must be
enabled. See :ref:`guide-routing` for more
information.
@@ -446,7 +445,7 @@ def apply_async(self, args=None, kwargs=None, task_id=None, producer=None,
to use. Can be one of ``zlib``, ``bzip2``,
or any custom compression methods registered with
:func:`kombu.compression.register`. Defaults to
the :setting:`CELERY_MESSAGE_COMPRESSION`
the :setting:`task_compression`
setting.
:keyword link: A single, or a list of tasks to apply if the
task exits successfully.
@@ -467,14 +466,14 @@ def apply_async(self, args=None, kwargs=None, task_id=None, producer=None,
task (a :class:`dict`)
:rtype :class:`celery.result.AsyncResult`: if
:setting:`CELERY_ALWAYS_EAGER` is not set, otherwise
:setting:`task_always_eager` is not set, otherwise
:class:`celery.result.EagerResult`:
Also supports all keyword arguments supported by
:meth:`kombu.Producer.publish`.
.. note::
If the :setting:`CELERY_ALWAYS_EAGER` setting is set, it will
If the :setting:`task_always_eager` setting is set, it will
be replaced by a local :func:`apply` call instead.
"""
@@ -486,7 +485,7 @@ def apply_async(self, args=None, kwargs=None, task_id=None, producer=None,
check_arguments(*(args or ()), **(kwargs or {}))
app = self._get_app()
if app.conf.CELERY_ALWAYS_EAGER:
if app.conf.task_always_eager:
return self.apply(args, kwargs, task_id=task_id or uuid(),
link=link, link_error=link_error, **options)
# add 'self' if this is a "task_method".
@@ -670,7 +669,7 @@ def apply(self, args=None, kwargs=None,
:param args: positional arguments passed on to the task.
:param kwargs: keyword arguments passed on to the task.
:keyword throw: Re-raise task exceptions. Defaults to
the :setting:`CELERY_EAGER_PROPAGATES_EXCEPTIONS`
the :setting:`task_eager_propagates_exceptions`
setting.
:rtype :class:`celery.result.EagerResult`:
@@ -687,7 +686,7 @@ def apply(self, args=None, kwargs=None,
kwargs = kwargs or {}
task_id = options.get('task_id') or uuid()
retries = options.get('retries', 0)
throw = app.either('CELERY_EAGER_PROPAGATES_EXCEPTIONS',
throw = app.either('task_eager_propagates_exceptions',
options.pop('throw', None))
# Make sure we get the task instance, not class.
View
4 celery/app/trace.py
@@ -464,7 +464,7 @@ def _trace_task_ret(name, uuid, request, body, content_type,
app = app or current_app._get_current_object()
embed = None
if content_type:
accept = prepare_accept_content(app.conf.CELERY_ACCEPT_CONTENT)
accept = prepare_accept_content(app.conf.accept_content)
args, kwargs, embed = loads(
body, content_type, content_encoding, accept=accept,
)
@@ -539,7 +539,7 @@ def setup_worker_optimizations(app, hostname=None):
# set fast shortcut to task registry
_localized[:] = [
app._tasks,
prepare_accept_content(app.conf.CELERY_ACCEPT_CONTENT),
prepare_accept_content(app.conf.accept_content),
hostname,
]
View
135 celery/app/utils.py
@@ -12,18 +12,23 @@
import platform as _platform
import re
from collections import Mapping
from collections import Mapping, namedtuple
from copy import deepcopy
from types import ModuleType
from kombu.utils.url import maybe_sanitize_url
from celery.datastructures import ConfigurationView
from celery.five import items, string_t, values
from celery.exceptions import ImproperlyConfigured
from celery.five import items, keys, string_t, values
from celery.platforms import pyimplementation
from celery.utils.text import pretty
from celery.utils.imports import import_from_cwd, symbol_by_name, qualname
from .defaults import find
from .defaults import (
_TO_NEW_KEY, _TO_OLD_KEY, _OLD_DEFAULTS, _OLD_SETTING_KEYS,
DEFAULTS, SETTING_KEYS, find,
)
__all__ = ['Settings', 'appstr', 'bugreport',
'filter_hidden_settings', 'find_app']
@@ -44,6 +49,28 @@
re.IGNORECASE,
)
E_MIX_OLD_INTO_NEW = """
Cannot mix new and old setting keys, please rename the
following settings to the new format:
{renames}
"""
E_MIX_NEW_INTO_OLD = """
Cannot mix new setting names with old setting names, please
rename the following settings to use the old format:
{renames}
Or change all of the settings to use the new format :)
"""
FMT_REPLACE_SETTING = '{replace:<36} -> {with_}'
def appstr(app):
"""String used in __repr__ etc, to id app instances."""
@@ -60,28 +87,14 @@ class Settings(ConfigurationView):
"""
@property
def CELERY_RESULT_BACKEND(self):
return self.first('CELERY_RESULT_BACKEND', 'CELERY_BACKEND')
@property
def BROKER_TRANSPORT(self):
return self.first('BROKER_TRANSPORT',
'BROKER_BACKEND', 'CARROT_BACKEND')
@property
def BROKER_BACKEND(self):
"""Deprecated compat alias to :attr:`BROKER_TRANSPORT`."""
return self.BROKER_TRANSPORT
@property
def BROKER_URL(self):
def broker_url(self):
return (os.environ.get('CELERY_BROKER_URL') or
self.first('BROKER_URL', 'BROKER_HOST'))
self.first('broker_url', 'broker_host'))
@property
def CELERY_TIMEZONE(self):
def timezone(self):
# this way we also support django's time zone.
return self.first('CELERY_TIMEZONE', 'TIME_ZONE')
return self.first('timezone', 'time_zone')
def without_defaults(self):
"""Return the current configuration, but without defaults."""
@@ -91,18 +104,18 @@ def without_defaults(self):
def value_set_for(self, key):
return key in self.without_defaults()
def find_option(self, name, namespace='celery'):
def find_option(self, name, namespace=''):
"""Search for option by name.
Will return ``(namespace, key, type)`` tuple, e.g.::
>>> from proj.celery import app
>>> app.conf.find_option('disable_rate_limits')
('CELERY', 'DISABLE_RATE_LIMITS',
('worker', 'prefetch_multiplier',
<Option: type->bool default->False>))
:param name: Name of option, cannot be partial.
:keyword namespace: Preferred namespace (``CELERY`` by default).
:keyword namespace: Preferred namespace (``None`` by default).
"""
return find(name, namespace)
@@ -117,7 +130,7 @@ def get_by_parts(self, *parts):
Example::
>>> from proj.celery import app
>>> app.conf.get_by_parts('CELERY', 'DISABLE_RATE_LIMITS')
>>> app.conf.get_by_parts('worker', 'disable_rate_limits')
False
"""
@@ -139,6 +152,72 @@ def humanize(self, with_defaults=False, censored=True):
for key, value in items(self.table(with_defaults, censored)))
def _new_key_to_old(key, convert=_TO_OLD_KEY.get):
return convert(key, key)
def _old_key_to_new(key, convert=_TO_NEW_KEY.get):
return convert(key, key)
_settings_info_t = namedtuple('settings_info_t', (
'defaults', 'convert', 'key_t', 'mix_error',
))
_settings_info = _settings_info_t(
DEFAULTS, _TO_NEW_KEY, _old_key_to_new, E_MIX_OLD_INTO_NEW,
)
_old_settings_info = _settings_info_t(
_OLD_DEFAULTS, _TO_OLD_KEY, _new_key_to_old, E_MIX_NEW_INTO_OLD,
)
def detect_settings(conf, preconf={}, ignore_keys=set(), prefix=None,
all_keys=SETTING_KEYS, old_keys=_OLD_SETTING_KEYS):
source = conf
if conf is None:
source, conf = preconf, {}
have = set(keys(source)) - ignore_keys
is_in_new = have.intersection(all_keys)
is_in_old = have.intersection(old_keys)
if is_in_new:
# have new setting names
info, left = _settings_info, is_in_old
if is_in_old and len(is_in_old) > len(is_in_new):
# Majority of the settings are old.
info, left = _old_settings_info, is_in_new
elif is_in_old:
print('IS IN OLD: %r' % (is_in_old, ))
# have old setting names, or a majority of the names are old.
info, left = _old_settings_info, is_in_new
if is_in_new and len(is_in_new) > len(is_in_old):
# Majority of the settings are new
info, left = _settings_info, is_in_old
else:
# no settings, just use new format.
info, left = _settings_info, is_in_old
if prefix:
# always use new format if prefix is used.
info, left = _settings_info, set()
# only raise error for keys that the user did not provide two keys
# for (e.g. both ``result_expires`` and ``CELERY_TASK_RESULT_EXPIRES``).
really_left = {key for key in left if info.convert[key] not in have}
if really_left:
# user is mixing old/new, or new/old settings, give renaming
# suggestions.
raise ImproperlyConfigured(info.mix_error.format(renames='\n'.join(
FMT_REPLACE_SETTING.format(replace=key, with_=info.convert[key])
for key in sorted(really_left)
)))
preconf = {info.convert.get(k, k): v for k, v in items(preconf)}
defaults = dict(deepcopy(info.defaults), **preconf)
return Settings(preconf, [conf, defaults], info.key_t, prefix=prefix)
class AppPickler(object):
"""Old application pickler/unpickler (< 3.1)."""
@@ -185,10 +264,10 @@ def maybe_censor(key, value, mask='*' * 8):
if isinstance(key, string_t):
if HIDDEN_SETTINGS.search(key):
return mask
elif 'BROKER_URL' in key.upper():
elif 'broker_url' in key.lower():
from kombu import Connection
return Connection(value).as_uri(mask=mask)
elif key.upper() in ('CELERY_RESULT_BACKEND', 'CELERY_BACKEND'):
elif 'backend' in key.lower():
return maybe_sanitize_url(value, mask=mask)
return value
@@ -220,7 +299,7 @@ def bugreport(app):
py_v=_platform.python_version(),
driver_v=driver_v,
transport=transport,
results=app.conf.CELERY_RESULT_BACKEND or 'disabled',
results=app.conf.result_backend or 'disabled',
human_settings=app.conf.humanize(),
loader=qualname(app.loader.__class__),
)
View
18 celery/apps/beat.py
@@ -48,15 +48,16 @@ def __init__(self, max_interval=None, app=None,
redirect_stdouts_level=None, **kwargs):
"""Starts the beat task scheduler."""
self.app = app = app or self.app
either = self.app.either
self.loglevel = loglevel
self.logfile = logfile
self.schedule = self._getopt('schedule_filename', schedule)
self.scheduler_cls = self._getopt('scheduler', scheduler_cls)
self.redirect_stdouts = self._getopt(
'redirect_stdouts', redirect_stdouts,
self.schedule = either('beat_schedule_filename', schedule)
self.scheduler_cls = either('beat_scheduler', scheduler_cls)
self.redirect_stdouts = either(
'worker_redirect_stdouts', redirect_stdouts,
)
self.redirect_stdouts_level = self._getopt(
'redirect_stdouts_level', redirect_stdouts_level,
self.redirect_stdouts_level = either(
'worker_redirect_stdouts_level', redirect_stdouts_level,
)
self.max_interval = max_interval
@@ -71,11 +72,6 @@ def __init__(self, max_interval=None, app=None,
if not isinstance(self.loglevel, numbers.Integral):
self.loglevel = LOG_LEVELS[self.loglevel.upper()]
def _getopt(self, key, value):
if value is not None:
return value
return self.app.conf.find_value_for_key(key, namespace='celerybeat')
def run(self):
print(str(self.colored.cyan(
'celery beat v{0} is starting.'.format(VERSION_BANNER))))
View
17 celery/apps/worker.py
@@ -57,7 +57,7 @@
warning and to be sure that everything will continue working
when you upgrade to Celery 4.0::
CELERY_ACCEPT_CONTENT = ['pickle', 'json', 'msgpack', 'yaml']
accept_content = ['pickle', 'json', 'msgpack', 'yaml']
You must only enable the serializers that you will actually use.
@@ -120,16 +120,16 @@ def on_before_init(self, **kwargs):
sender=self.hostname, instance=self,
conf=self.app.conf, options=kwargs,
)
check_privileges(self.app.conf.CELERY_ACCEPT_CONTENT)
check_privileges(self.app.conf.accept_content)
def on_after_init(self, purge=False, no_color=None,
redirect_stdouts=None, redirect_stdouts_level=None,
**kwargs):
self.redirect_stdouts = self._getopt(
'redirect_stdouts', redirect_stdouts,
self.redirect_stdouts = self.app.either(
'worker_redirect_stdouts', redirect_stdouts,
)
self.redirect_stdouts_level = self._getopt(
'redirect_stdouts_level', redirect_stdouts_level,
self.redirect_stdouts_level = self.app.either(
'worker_redirect_stdouts_level', redirect_stdouts_level,
)
super(Worker, self).setup_defaults(**kwargs)
self.purge = purge
@@ -158,7 +158,7 @@ def on_start(self):
sender=self.hostname, instance=self, conf=self.app.conf,
)
if not self.app.conf.value_set_for('CELERY_ACCEPT_CONTENT'):
if not self.app.conf.value_set_for('accept_content'):
warnings.warn(CDeprecationWarning(W_PICKLE_DEPRECATED))
if self.purge:
@@ -229,7 +229,7 @@ def startup_info(self):
version=VERSION_BANNER,
conninfo=self.app.connection().as_uri(),
results=maybe_sanitize_url(
self.app.conf.CELERY_RESULT_BACKEND or 'disabled',
self.app.conf.result_backend or 'disabled',
),
concurrency=concurrency,
platform=safe_str(_platform.platform()),
@@ -281,7 +281,6 @@ def set_process_status(self, info):
def _shutdown_handler(worker, sig='TERM', how='Warm',
exc=WorkerShutdown, callback=None, exitcode=EX_OK):
def _handle_request(*args):
with in_sighandler():
from celery.worker import state
View
6 celery/backends/amqp.py
@@ -73,12 +73,12 @@ def __init__(self, app, connection=None, exchange=None, exchange_type=None,
self._connection = connection
self.persistent = self.prepare_persistent(persistent)
self.delivery_mode = 2 if self.persistent else 1
exchange = exchange or conf.CELERY_RESULT_EXCHANGE
exchange_type = exchange_type or conf.CELERY_RESULT_EXCHANGE_TYPE
exchange = exchange or conf.result_exchange
exchange_type = exchange_type or conf.result_exchange_type
self.exchange = self._create_exchange(
exchange, exchange_type, self.delivery_mode,
)
self.serializer = serializer or conf.CELERY_RESULT_SERIALIZER
self.serializer = serializer or conf.result_serializer
self.auto_delete = auto_delete
self.queue_arguments = dictfilter({
'x-expires': maybe_s_to_ms(self.expires),
View
12 celery/backends/base.py
@@ -96,16 +96,16 @@ def __init__(self, app,
expires=None, expires_type=None, **kwargs):
self.app = app
conf = self.app.conf
self.serializer = serializer or conf.CELERY_RESULT_SERIALIZER
self.serializer = serializer or conf.result_serializer
(self.content_type,
self.content_encoding,
self.encoder) = serializer_registry._encoders[self.serializer]
cmax = max_cached_results or conf.CELERY_MAX_CACHED_RESULTS
cmax = max_cached_results or conf.result_cache_max
self._cache = _nulldict() if cmax == -1 else LRUCache(limit=cmax)
self.expires = self.prepare_expires(expires, expires_type)
self.accept = prepare_accept_content(
conf.CELERY_ACCEPT_CONTENT if accept is None else accept,
conf.accept_content if accept is None else accept,
)
def mark_as_started(self, task_id, **meta):
@@ -242,7 +242,7 @@ def wait_for(self, task_id,
def prepare_expires(self, value, type=None):
if value is None:
value = self.app.conf.CELERY_TASK_RESULT_EXPIRES
value = self.app.conf.result_expires
if isinstance(value, timedelta):
value = value.total_seconds()
if value is not None and type:
@@ -252,7 +252,7 @@ def prepare_expires(self, value, type=None):
def prepare_persistent(self, enabled=None):
if enabled is not None:
return enabled
p = self.app.conf.CELERY_RESULT_PERSISTENT
p = self.app.conf.result_persistent
return self.persistent if p is None else p
def encode_result(self, result, status):
@@ -558,7 +558,7 @@ def on_chord_part_return(self, request, state, result, propagate=None):
return
app = self.app
if propagate is None:
propagate = app.conf.CELERY_CHORD_PROPAGATES
propagate = app.conf.chord_propagates
gid = request.group
if not gid:
return
View
4 celery/backends/cache.py
@@ -101,10 +101,10 @@ def __init__(self, app, expires=None, backend=None,
options={}, url=None, **kwargs):
super(CacheBackend, self).__init__(app, **kwargs)
self.options = dict(self.app.conf.CELERY_CACHE_BACKEND_OPTIONS,
self.options = dict(self.app.conf.cache_backend_options,
**options)
self.backend = url or backend or self.app.conf.CELERY_CACHE_BACKEND
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(';')
View
16 celery/backends/cassandra.py
@@ -59,7 +59,7 @@ def __init__(self, servers=None, keyspace=None, column_family=None,
"""Initialize Cassandra backend.
Raises :class:`celery.exceptions.ImproperlyConfigured` if
the :setting:`CASSANDRA_SERVERS` setting is not set.
the :setting:`cassandra_servers` setting is not set.
"""
super(CassandraBackend, self).__init__(**kwargs)
@@ -71,21 +71,21 @@ def __init__(self, servers=None, keyspace=None, column_family=None,
conf = self.app.conf
self.servers = (servers or
conf.get('CASSANDRA_SERVERS') or
conf.get('cassandra_servers') or
self.servers)
self.keyspace = (keyspace or
conf.get('CASSANDRA_KEYSPACE') or
conf.get('cassandra_keyspace') or
self.keyspace)
self.column_family = (column_family or
conf.get('CASSANDRA_COLUMN_FAMILY') or
conf.get('cassandra_column_family') or
self.column_family)
self.cassandra_options = dict(conf.get('CASSANDRA_OPTIONS') or {},
self.cassandra_options = dict(conf.get('cassandra_options') or {},
**cassandra_options or {})
self.detailed_mode = (detailed_mode or
conf.get('CASSANDRA_DETAILED_MODE') or
conf.get('cassandra_detailed_mode') or
self.detailed_mode)
read_cons = conf.get('CASSANDRA_READ_CONSISTENCY') or 'LOCAL_QUORUM'
write_cons = conf.get('CASSANDRA_WRITE_CONSISTENCY') or 'LOCAL_QUORUM'
read_cons = conf.get('cassandra_read_consistency') or 'LOCAL_QUORUM'
write_cons = conf.get('cassandra_write_consistency') or 'LOCAL_QUORUM'
try:
self.read_consistency = getattr(pycassa.ConsistencyLevel,
read_cons)
View
2 celery/backends/couchbase.py
@@ -63,7 +63,7 @@ def __init__(self, url=None, *args, **kwargs):
_, uhost, uport, uname, upass, ubucket, _ = _parse_url(url)
ubucket = ubucket.strip('/') if ubucket else None
config = self.app.conf.get('CELERY_COUCHBASE_BACKEND_SETTINGS', None)
config = self.app.conf.get('couchbase_backend_settings', None)
if config is not None:
if not isinstance(config, dict):
raise ImproperlyConfigured(
View
12 celery/backends/database/__init__.py
@@ -80,23 +80,23 @@ def __init__(self, dburi=None, engine_options=None, url=None, **kwargs):
expires_type=maybe_timedelta, **kwargs
)
conf = self.app.conf
self.dburi = url or dburi or conf.CELERY_RESULT_DBURI
self.dburi = url or dburi or conf.sqlalchemy_dburi
self.engine_options = dict(
engine_options or {},
**conf.CELERY_RESULT_ENGINE_OPTIONS or {})
**conf.sqlalchemy_engine_options or {})
self.short_lived_sessions = kwargs.get(
'short_lived_sessions',
conf.CELERY_RESULT_DB_SHORT_LIVED_SESSIONS,
conf.sqlalchemy_short_lived_sessions,
)
tablenames = conf.CELERY_RESULT_DB_TABLENAMES or {}
tablenames = conf.sqlalchemy_table_names or {}
Task.__table__.name = tablenames.get('task', 'celery_taskmeta')
TaskSet.__table__.name = tablenames.get('group', 'celery_tasksetmeta')
if not self.dburi:
raise ImproperlyConfigured(
'Missing connection string! Do you have '
'CELERY_RESULT_DBURI set to a real value?')
'Missing connection string! Do you have the'
' sqlalchemy_dburi setting set to a real value?')
def ResultSession(self, session_manager=SessionManager()):
return session_manager.session_factory(
View
2 celery/backends/mongodb.py
@@ -98,7 +98,7 @@ def __init__(self, app=None, url=None, **kwargs):
self.options.update(uri_data['options'])
# update conf with specific settings
config = self.app.conf.get('CELERY_MONGODB_BACKEND_SETTINGS')
config = self.app.conf.get('mongodb_backend_settings')
if config is not None:
if not isinstance(config, dict):
raise ImproperlyConfigured(
View
16 celery/backends/new_cassandra.py
@@ -83,7 +83,7 @@ def __init__(self, servers=None, keyspace=None, table=None, entry_ttl=None,
"""Initialize Cassandra backend.
Raises :class:`celery.exceptions.ImproperlyConfigured` if
the :setting:`CASSANDRA_SERVERS` setting is not set.
the :setting:`cassandra_servers` setting is not set.
"""
super(CassandraBackend, self).__init__(**kwargs)
@@ -93,24 +93,24 @@ def __init__(self, servers=None, keyspace=None, table=None, entry_ttl=None,
conf = self.app.conf
self.servers = (servers or
conf.get('CASSANDRA_SERVERS', None))
conf.get('cassandra_servers', None))
self.port = (port or
conf.get('CASSANDRA_PORT', None))
conf.get('cassandra_port', None))
self.keyspace = (keyspace or
conf.get('CASSANDRA_KEYSPACE', None))
conf.get('cassandra_keyspace', None))
self.table = (table or
conf.get('CASSANDRA_TABLE', None))
conf.get('cassandra_table', None))
if not self.servers or not self.keyspace or not self.table:
raise ImproperlyConfigured('Cassandra backend not configured.')
expires = (entry_ttl or conf.get('CASSANDRA_ENTRY_TTL', None))
expires = (entry_ttl or conf.get('cassandra_entry_ttl', None))
self.cqlexpires = (Q_EXPIRES.format(expires)
if expires is not None else '')
read_cons = conf.get('CASSANDRA_READ_CONSISTENCY') or 'LOCAL_QUORUM'
write_cons = conf.get('CASSANDRA_WRITE_CONSISTENCY') or 'LOCAL_QUORUM'
read_cons = conf.get('cassandra_read_consistency') or 'LOCAL_QUORUM'
write_cons = conf.get('cassandra_write_consistency') or 'LOCAL_QUORUM'
self.read_consistency = getattr(
cassandra.ConsistencyLevel, read_cons,
View
23 celery/backends/redis.py
@@ -60,32 +60,27 @@ def __init__(self, host=None, port=None, db=None, password=None,
max_connections=None, url=None,
connection_pool=None, new_join=False, **kwargs):
super(RedisBackend, self).__init__(expires_type=int, **kwargs)
conf = self.app.conf
_get = self.app.conf.get
if self.redis is None:
raise ImproperlyConfigured(REDIS_MISSING)
# 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
max_connections or
_get('redis_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'),
'socket_timeout': _get('SOCKET_TIMEOUT'),
'host': _get('redis_host') or 'localhost',
'port': _get('redis_port') or 6379,
'db': _get('redis_db') or 0,
'password': _get('redis_password'),
'socket_timeout': _get('redis_socket_timeout'),
'max_connections': self.max_connections,
}
if url:
View
2 celery/backends/riak.py
@@ -85,7 +85,7 @@ def __init__(self, host=None, port=None, bucket_name=None, protocol=None,
if ubucket:
ubucket = ubucket.strip('/')
config = self.app.conf.get('CELERY_RIAK_BACKEND_SETTINGS', None)
config = self.app.conf.get('riak_backend_settings', None)
if config is not None:
if not isinstance(config, dict):
raise ImproperlyConfigured(
View
18 celery/beat.py
@@ -186,19 +186,19 @@ def __init__(self, app, schedule=None, max_interval=None,
self.app = app
self.data = maybe_evaluate({} if schedule is None else schedule)
self.max_interval = (max_interval or
app.conf.CELERYBEAT_MAX_LOOP_INTERVAL or
app.conf.beat_max_loop_interval or
self.max_interval)
self.Producer = Producer or app.amqp.Producer
self._heap = None
self.sync_every_tasks = (
app.conf.CELERYBEAT_SYNC_EVERY if sync_every_tasks is None
app.conf.beat_sync_every if sync_every_tasks is None
else sync_every_tasks)
if not lazy:
self.setup_schedule()
def install_default_entries(self, data):
entries = {}
if self.app.conf.CELERY_TASK_RESULT_EXPIRES and \
if self.app.conf.result_expires and \
not self.app.backend.supports_autoexpire:
if 'celery.backend_cleanup' not in data:
entries['celery.backend_cleanup'] = {
@@ -363,7 +363,7 @@ def _error_handler(exc, interval):
'Trying again in %s seconds...', exc, interval)
return self.connection.ensure_connection(
_error_handler, self.app.conf.BROKER_CONNECTION_MAX_RETRIES
_error_handler, self.app.conf.broker_connection_max_retries
)
def get_schedule(self):
@@ -438,20 +438,20 @@ def setup_schedule(self):
self._store.clear() # remove schedule at 3.0.9 upgrade
break
tz = self.app.conf.CELERY_TIMEZONE
tz = self.app.conf.timezone
stored_tz = self._store.get('tz')
if stored_tz is not None and stored_tz != tz:
warning('Reset: Timezone changed from %r to %r', stored_tz, tz)
self._store.clear() # Timezone changed, reset db!
utc = self.app.conf.CELERY_ENABLE_UTC
utc = self.app.conf.enable_utc
stored_utc = self._store.get('utc_enabled')
if stored_utc is not None and stored_utc != utc:
choices = {True: 'enabled', False: 'disabled'}
warning('Reset: UTC changed from %s to %s',
choices[stored_utc], choices[utc])
self._store.clear() # UTC setting changed, reset db!
entries = self._store.setdefault('entries', {})
self.merge_inplace(self.app.conf.CELERYBEAT_SCHEDULE)
self.merge_inplace(self.app.conf.beat_schedule)
self.install_default_entries(self.schedule)
self._store.update(__version__=__version__, tz=tz, utc_enabled=utc)
self.sync()
@@ -485,10 +485,10 @@ def __init__(self, app, max_interval=None, schedule_filename=None,
scheduler_cls=None):
self.app = app
self.max_interval = (max_interval or
app.conf.CELERYBEAT_MAX_LOOP_INTERVAL)
app.conf.beat_max_loop_interval)
self.scheduler_cls = scheduler_cls or self.scheduler_cls
self.schedule_filename = (
schedule_filename or app.conf.CELERYBEAT_SCHEDULE_FILENAME)
schedule_filename or app.conf.beat_schedule_filename)
self._is_shutdown = Event()
self._is_stopped = Event()
View
2 celery/bin/base.py
@@ -219,7 +219,7 @@ class Command(object):
enable_config_from_cmdline = False
#: Default configuration namespace.
namespace = 'celery'
namespace = None
#: Text to print at end of --help
epilog = None
View
2 celery/bin/beat.py
@@ -84,7 +84,7 @@ def get_options(self):
return (
(Option('--detach', action='store_true'),
Option('-s', '--schedule',
default=c.CELERYBEAT_SCHEDULE_FILENAME),
default=c.beat_schedule_filename),
Option('--max-interval', type='float'),
Option('-S', '--scheduler', dest='scheduler_cls'),
Option('-l', '--loglevel', default='WARN')) +
View
1 celery/bin/celery.py
@@ -660,7 +660,6 @@ def run(self, *args, **kwargs):
class CeleryCommand(Command):
namespace = 'celery'
ext_fmt = '{self.namespace}.commands'
commands = {
'amqp': amqp,
View
2 celery/bin/graph.py
@@ -156,7 +156,7 @@ def maybe_abbr(l, name, max=Wmax):
threads.append(reply['pool']['max-concurrency'])
wlen = len(workers)
backend = args.get('backend', self.app.conf.CELERY_RESULT_BACKEND)
backend = args.get('backend', self.app.conf.result_backend)
threads_for = {}
workers = maybe_abbr(workers, 'Worker')
if Wmax and wlen > Wmax:
View
24 celery/bin/worker.py
@@ -173,7 +173,7 @@ class worker(Command):
celery worker --autoscale=10,0
"""
doc = __MODULE_DOC__ # parse help from this too
namespace = 'celeryd'
namespace = 'worker'
enable_config_from_cmdline = True
supports_args = False
@@ -200,7 +200,7 @@ def run(self, hostname=None, pool_cls=None, app=None, uid=None, gid=None,
# Pools like eventlet/gevent needs to patch libs as early
# as possible.
pool_cls = (concurrency.get_implementation(pool_cls) or
self.app.conf.CELERYD_POOL)
self.app.conf.worker_pool)
if self.app.IS_WINDOWS and kwargs.get('beat'):
self.die('-B option does not work on Windows. '
'Please run celery beat as a separate service.')
@@ -231,29 +231,29 @@ def get_options(self):
conf = self.app.conf
return (
Option('-c', '--concurrency',
default=conf.CELERYD_CONCURRENCY, type='int'),
Option('-P', '--pool', default=conf.CELERYD_POOL, dest='pool_cls'),
default=conf.worker_concurrency, type='int'),
Option('-P', '--pool', default=conf.worker_pool, dest='pool_cls'),
Option('--purge', '--discard', default=False, action='store_true'),
Option('-l', '--loglevel', default='WARN'),
Option('-n', '--hostname'),
Option('-B', '--beat', action='store_true'),
Option('-s', '--schedule', dest='schedule_filename',
default=conf.CELERYBEAT_SCHEDULE_FILENAME),
default=conf.beat_schedule_filename),
Option('--scheduler', dest='scheduler_cls'),
Option('-S', '--statedb',
default=conf.CELERYD_STATE_DB, dest='state_db'),
Option('-E', '--events', default=conf.CELERY_SEND_EVENTS,
default=conf.worker_state_db, dest='state_db'),
Option('-E', '--events', default=conf.worker_send_events,
action='store_true', dest='send_events'),
Option('--time-limit', type='float', dest='task_time_limit',
default=conf.CELERYD_TASK_TIME_LIMIT),
default=conf.task_time_limit),
Option('--soft-time-limit', dest='task_soft_time_limit',
default=conf.CELERYD_TASK_SOFT_TIME_LIMIT, type='float'),
default=conf.task_soft_time_limit, type='float'),
Option('--maxtasksperchild', dest='max_tasks_per_child',
default=conf.CELERYD_MAX_TASKS_PER_CHILD, type='int'),
default=conf.worker_max_tasks_per_child, type='int'),
Option('--prefetch-multiplier', dest='prefetch_multiplier',
default=conf.CELERYD_PREFETCH_MULTIPLIER, type='int'),
default=conf.worker_prefetch_multiplier, type='int'),
Option('--maxmemperchild', dest='max_memory_per_child',
default=conf.CELERYD_MAX_MEMORY_PER_CHILD, type='int'),
default=conf.worker_max_memory_per_child, type='int'),
Option('--queues', '-Q', default=[]),
Option('--exclude-queues', '-X', default=[]),
Option('--include', '-I', default=[]),
View
8 celery/canvas.py
@@ -393,7 +393,7 @@ def __call__(self, *args, **kwargs):
def apply_async(self, args=(), kwargs={}, **options):
# python is best at unpacking kwargs, so .run is here to do that.
app = self.app
if app.conf.CELERY_ALWAYS_EAGER:
if app.conf.task_always_eager:
return self.apply(args, kwargs, **options)
return self.run(args, kwargs, app=app, **(
dict(self.options, **options) if options else self.options))
@@ -688,7 +688,7 @@ def _freeze_gid(self, options):
def apply_async(self, args=(), kwargs=None, add_to_parent=True,
producer=None, **options):
app = self.app
if app.conf.CELERY_ALWAYS_EAGER:
if app.conf.task_always_eager:
return self.apply(args, kwargs, **options)
if not self.tasks:
return self.freeze()
@@ -846,7 +846,7 @@ def apply_async(self, args=(), kwargs={}, task_id=None,
app = self._get_app(body)
tasks = (self.tasks.clone() if isinstance(self.tasks, group)
else group(self.tasks))
if app.conf.CELERY_ALWAYS_EAGER:
if app.conf.task_always_eager:
return self.apply((), kwargs,
body=body, task_id=task_id, **options)
return self.run(tasks, body, args, task_id=task_id, **options)
@@ -875,7 +875,7 @@ def run(self, header, body, partial_args, app=None, interval=None,
countdown=1, max_retries=None, propagate=None, eager=False,
task_id=None, **options):
app = app or self._get_app(body)
propagate = (app.conf.CELERY_CHORD_PROPAGATES
propagate = (app.conf.chord_propagates
if propagate is None else propagate)
group_id = uuid()
root_id = body.options.get('root_id')
View
2 celery/contrib/batches.py
@@ -8,7 +8,7 @@
.. warning::
For this to work you have to set
:setting:`CELERYD_PREFETCH_MULTIPLIER` to zero, or some value where
:setting:`worker_prefetch_multiplier` to zero, or some value where
the final multiplied value is higher than ``flush_every``.
In the future we hope to add the ability to direct batching tasks
View
2 celery/contrib/migrate.py
@@ -141,7 +141,7 @@ def move(predicate, connection=None, exchange=None, routing_key=None,
:keyword connection: Custom connection to use.
:keyword source: Optional list of source queues to use instead of the
default (which is the queues in :setting:`CELERY_QUEUES`).
default (which is the queues in :setting:`task_queues`).
This list can also contain new :class:`~kombu.entity.Queue` instances.
:keyword exchange: Default destination exchange.
:keyword routing_key: Default destination routing key.
View
52 celery/datastructures.py
@@ -387,11 +387,8 @@ def get(self, key, default=None):
return default
def setdefault(self, key, default):
try:
return self[key]
except KeyError:
if key not in self:
self[key] = default
return default
def __getitem__(self, key):
try:
@@ -451,20 +448,35 @@ class ConfigurationView(AttributeDictMixin):
:param defaults: List of dicts containing the default configuration.
"""
key_t = None
changes = None
defaults = None
_order = None
def __init__(self, changes, defaults):
self.__dict__.update(changes=changes, defaults=defaults,
_order=[changes] + defaults)
def __init__(self, changes, defaults=None, key_t=None, prefix=None):
defaults = [] if defaults is None else defaults
self.__dict__.update(
changes=changes,
defaults=defaults,
key_t=key_t,
_order=[changes] + defaults,
prefix=prefix,
)
def _key(self, key):
if self.prefix:
key = self.prefix + key
if self.prefix.isupper():
key = key.upper()
return self.key_t(key) if self.key_t is not None else key
def add_defaults(self, d):
d = force_mapping(d)
self.defaults.insert(0, d)
self._order.insert(1, d)
def __getitem__(self, key):
key = self._key(key)
for d in self._order:
try:
return d[key]
@@ -473,14 +485,14 @@ def __getitem__(self, key):
raise KeyError(key)
def __setitem__(self, key, value):
self.changes[key] = value
self.changes[self._key(key)] = value
def first(self, *keys):
return first(None, (self.get(key) for key in keys))
return first(None, (self.get(self._key(key)) for key in keys))
def get(self, key, default=None):
try:
return self[key]
return self[self._key(key)]
except KeyError:
return default
@@ -489,16 +501,15 @@ def clear(self):
self.changes.clear()
def setdefault(self, key, default):
try:
return self[key]
except KeyError:
key = self._key(key)
if key not in self:
self[key] = default
return default
def update(self, *args, **kwargs):
return self.changes.update(*args, **kwargs)
def __contains__(self, key):
key = self._key(key)
return any(key in m for m in self._order)
def __bool__(self):
@@ -521,8 +532,19 @@ def _iter(self, op):
# changes takes precedence.
return chain(*[op(d) for d in reversed(self._order)])
def swap_with(self, other):
changes = other.__dict__['changes']
defaults = other.__dict__['defaults']
self.__dict__.update(
changes=changes,
defaults=defaults,
key_t=other.__dict__['key_t'],
prefix=other.__dict__['prefix'],
_order=[changes] + defaults
)
def _iterate_keys(self):
return uniq(self._iter(lambda d: d))
return uniq(self._iter(lambda d: d.keys()))
iterkeys = _iterate_keys
def _iterate_items(self):
View
10 celery/events/__init__.py
@@ -4,7 +4,7 @@
~~~~~~~~~~~~~
Events is a stream of messages sent for certain actions occurring
in the worker (and clients if :setting:`CELERY_SEND_TASK_SENT_EVENT`
in the worker (and clients if :setting:`task_send_sent_event`
is enabled), used for monitoring purposes.
"""
@@ -130,7 +130,7 @@ def __init__(self, connection=None, hostname=None, enabled=True,
self.mutex = threading.Lock()
self.producer = None
self._outbound_buffer = deque()
self.serializer = serializer or self.app.conf.CELERY_EVENT_SERIALIZER
self.serializer = serializer or self.app.conf.event_serializer
self.on_enabled = set()
self.on_disabled = set()
self.groups = set(groups or [])
@@ -321,18 +321,18 @@ def __init__(self, channel, handlers=None, routing_key='#',
self.adjust_clock = self.clock.adjust
self.forward_clock = self.clock.forward
if accept is None:
accept = {self.app.conf.CELERY_EVENT_SERIALIZER, 'json'}
accept = {self.app.conf.event_serializer, 'json'}
self.accept = accept
def _get_queue_arguments(self, ttl=None, expires=None):
conf = self.app.conf
return dictfilter({
'x-message-ttl': maybe_s_to_ms(
ttl if ttl is not None else conf.CELERY_EVENT_QUEUE_TTL,
ttl if ttl is not None else conf.event_queue_ttl,
),
'x-expires': maybe_s_to_ms(
expires if expires is not None
else conf.CELERY_EVENT_QUEUE_EXPIRES,
else conf.event_queue_expires,
),
})
View
2 celery/events/cursesmon.py
@@ -511,7 +511,7 @@ def on_connection_error(exc, interval):
with app.connection() as conn:
try:
conn.ensure_connection(on_connection_error,
app.conf.BROKER_CONNECTION_MAX_RETRIES)
app.conf.broker_connection_max_retries)
recv = app.events.Receiver(conn, handlers={'*': state.event})
display.resetscreen()
display.init_screen()
View
4 celery/exceptions.py
@@ -120,7 +120,7 @@ class WorkerShutdown(SystemExit):
class QueueNotFound(KeyError):
"""Task routed to a queue not in CELERY_QUEUES."""
"""Task routed to a queue not in ``conf.queues``."""
class ImproperlyConfigured(ImportError):
@@ -155,7 +155,7 @@ class NotConfigured(CeleryWarning):
class AlwaysEagerIgnored(CeleryWarning):
"""send_task ignores CELERY_ALWAYS_EAGER option"""
"""send_task ignores :setting:`task_always_eager` option"""
class InvalidTaskError(CeleryError):
View
17 celery/loaders/base.py
@@ -40,6 +40,8 @@
Did you mean '{suggest}'?
"""
unconfigured = object()
class BaseLoader(object):
"""The base class for loaders.
@@ -65,7 +67,7 @@ class BaseLoader(object):
override_backends = {}
worker_initialized = False
_conf = None
_conf = unconfigured
def __init__(self, app, **kwargs):
self.app = app
@@ -117,8 +119,8 @@ def import_default_modules(self):
return [
self.import_task_module(m) for m in (
tuple(self.builtin_modules) +
tuple(maybe_list(self.app.conf.CELERY_IMPORTS)) +
tuple(maybe_list(self.app.conf.CELERY_INCLUDE))
tuple(maybe_list(self.app.conf.imports)) +
tuple(maybe_list(self.app.conf.include))
)
]
@@ -183,7 +185,7 @@ def cmdline_config_parser(
'list': 'json',
'dict': 'json'}):
from celery.app.defaults import Option, NAMESPACES
namespace = namespace.upper()
namespace = namespace.lower()
typemap = dict(Option.typemap, **extra_types)
def getarg(arg):
@@ -193,7 +195,7 @@ def getarg(arg):
# ## find key/value
# ns.key=value|ns_key=value (case insensitive)
key, value = arg.split('=', 1)
key = key.upper().replace('.', '_')
key = key.lower().replace('.', '_')
# ## find namespace.
# .key=value|_key=value expands to default namespace.
@@ -214,7 +216,7 @@ def getarg(arg):
value = typemap[type_](value)
else:
try:
value = NAMESPACES[ns][key].to_python(value)
value = NAMESPACES[ns.lower()][key].to_python(value)
except ValueError as exc:
# display key name in error message.
raise ValueError('{0!r}: {1}'.format(ns_key, exc))
@@ -244,7 +246,6 @@ def read_configuration(self, env='CELERY_CONFIG_MODULE'):
if custom_config:
usercfg = self._import_config_module(custom_config)
return DictAttribute(usercfg)
return {}
def autodiscover_tasks(self, packages, related_name='tasks'):
self.task_modules.update(
@@ -254,7 +255,7 @@ def autodiscover_tasks(self, packages, related_name='tasks'):
@property
def conf(self):
"""Loader configuration."""
if self._conf is None:
if self._conf is unconfigured:
self._conf = self.read_configuration()
return self._conf
View
4 celery/schedules.py
@@ -111,7 +111,7 @@ def is_due(self, last_run_at):
The next time to check is used to save energy/cpu cycles,
it does not need to be accurate but will influence the precision
of your schedule. You must also keep in mind
the value of :setting:`CELERYBEAT_MAX_LOOP_INTERVAL`,
the value of :setting:`beat_max_loop_interval`,
which decides the maximum number of seconds the scheduler can
sleep between re-checking the periodic task intervals. So if you
have a task that changes schedule at runtime then your next_run_at
@@ -172,7 +172,7 @@ def tz(self):
@cached_property
def utc_enabled(self):
return self.app.conf.CELERY_ENABLE_UTC
return self.app.conf.enable_utc
def to_local(self, dt):
if not self.utc_enabled:
View
14 celery/security/__init__.py
@@ -25,9 +25,9 @@
SETTING_MISSING = """\
Sorry, but you have to configure the
* CELERY_SECURITY_KEY
* CELERY_SECURITY_CERTIFICATE, and the
* CELERY_SECURITY_CERT_STORE
* security_key
* security_certificate, and the
* security_cert_storE
configuration settings to use the auth serializer.
Please see the configuration reference for more information.
@@ -46,17 +46,17 @@ def setup_security(allowed_serializers=None, key=None, cert=None, store=None,
_disable_insecure_serializers(allowed_serializers)
conf = app.conf
if conf.CELERY_TASK_SERIALIZER != 'auth':
if conf.task_serializer != 'auth':
return
try:
from OpenSSL import crypto # noqa
except ImportError:
raise ImproperlyConfigured(SSL_NOT_INSTALLED)
key = key or conf.CELERY_SECURITY_KEY
cert = cert or conf.CELERY_SECURITY_CERTIFICATE
store = store or conf.CELERY_SECURITY_CERT_STORE
key = key or conf.security_key
cert = cert or conf.security_certificate
store = store or conf.security_cert_store
if not (key and cert and store):
raise ImproperlyConfigured(SETTING_MISSING)
View
2 celery/states.py
@@ -128,7 +128,7 @@ def __le__(self, other):
PENDING = 'PENDING'
#: Task was received by a worker.
RECEIVED = 'RECEIVED'
#: Task was started by a worker (:setting:`CELERY_TRACK_STARTED`).
#: Task was started by a worker (:setting:`task_track_started`).
STARTED = 'STARTED'
#: Task succeeded
SUCCESS = 'SUCCESS'
View
10 celery/task/base.py
@@ -148,8 +148,8 @@ class Task(BaseTask):
disable_error_emails = False
from_config = BaseTask.from_config + (
('exchange_type', 'CELERY_DEFAULT_EXCHANGE_TYPE'),
('delivery_mode', 'CELERY_DEFAULT_DELIVERY_MODE'),
('exchange_type', 'task_default_exchange_type'),
('delivery_mode', 'task_default_delivery_mode'),
)
# In old Celery the @task decorator didn't exist, so one would create
@@ -244,7 +244,7 @@ def get_consumer(self, connection=None, queues=None, **kwargs):
class PeriodicTask(Task):
"""A periodic task is a task that adds itself to the
:setting:`CELERYBEAT_SCHEDULE` setting."""
:setting:`beat_schedule` setting."""
abstract = True
ignore_result = True
relative = False
@@ -260,7 +260,7 @@ def __init__(self):
@classmethod
def on_bound(cls, app):
app.conf.CELERYBEAT_SCHEDULE[cls.name] = {
app.conf.beat_schedule[cls.name] = {
'task': cls.name,
'schedule': cls.run_every,
'args': (),
@@ -276,5 +276,5 @@ def task(*args, **kwargs):
def periodic_task(*args, **options):
"""Deprecated decorator, please use :setting:`CELERYBEAT_SCHEDULE`."""
"""Deprecated decorator, please use :setting:`beat_schedule`."""
return task(**dict({'base': PeriodicTask}, **options))
View
2 celery/task/sets.py
@@ -53,7 +53,7 @@ def apply_async(self, connection=None, publisher=None, taskset_id=None):
"""Apply TaskSet."""
app = self.app
if app.conf.CELERY_ALWAYS_EAGER:
if app.conf.task_always_eager:
return self.apply(taskset_id=taskset_id)
with app.connection_or_acquire(connection) as conn:
View
6 celery/tests/app/test_amqp.py
@@ -11,7 +11,7 @@ class test_TaskConsumer(AppCase):
def test_accept_content(self):
with self.app.pool.acquire(block=True) as conn:
self.app.conf.CELERY_ACCEPT_CONTENT = ['application/json']
self.app.conf.accept_content = ['application/json']
self.assertEqual(
self.app.amqp.TaskConsumer(conn).accept,
{'application/json'},
@@ -25,7 +25,7 @@ def test_accept_content(self):
class test_ProducerPool(AppCase):
def test_setup_nolimit(self):
self.app.conf.BROKER_POOL_LIMIT = None
self.app.conf.broker_pool_limit = None
try:
delattr(self.app, '_pool')
except AttributeError:
@@ -43,7 +43,7 @@ def test_setup_nolimit(self):
r2 = pool.acquire()
def test_setup(self):
self.app.conf.BROKER_POOL_LIMIT = 2
self.app.conf.broker_pool_limit = 2
try:
delattr(self.app, '_pool')
except AttributeError:
View
210 celery/tests/app/test_app.py
@@ -15,8 +15,8 @@
from celery.app import base as _appbase
from celery.app import defaults
from celery.exceptions import ImproperlyConfigured
from celery.five import items
from celery.loaders.base import BaseLoader
from celery.five import items, keys
from celery.loaders.base import BaseLoader, unconfigured
from celery.platforms import pyimplementation
from celery.utils.serialization import pickle
@@ -225,7 +225,73 @@ def lazy_list():
@with_environ('CELERY_BROKER_URL', '')
def test_with_broker(self):
with self.Celery(broker='foo://baribaz') as app:
self.assertEqual(app.conf.BROKER_URL, 'foo://baribaz')
self.assertEqual(app.conf.broker_url, 'foo://baribaz')
def test_pending_configuration__setattr(self):
with self.Celery(broker='foo://bar') as app:
app.conf.task_default_delivery_mode = 44
app.conf.worker_agent = 'foo:Bar'
self.assertFalse(app.configured)
self.assertEqual(app.conf.worker_agent, 'foo:Bar')
self.assertEqual(app.conf.broker_url, 'foo://bar')
self.assertEqual(app._preconf['worker_agent'], 'foo:Bar')
self.assertTrue(app.configured)
reapp = pickle.loads(pickle.dumps(app))
self.assertEqual(reapp._preconf['worker_agent'], 'foo:Bar')
self.assertFalse(reapp.configured)
self.assertEqual(reapp.conf.worker_agent, 'foo:Bar')
self.assertTrue(reapp.configured)
self.assertEqual(reapp.conf.broker_url, 'foo://bar')
self.assertEqual(reapp._preconf['worker_agent'], 'foo:Bar')
def test_pending_configuration__update(self):
with self.Celery(broker='foo://bar') as app:
app.conf.update(
task_default_delivery_mode=44,
worker_agent='foo:Bar',
)
self.assertFalse(app.configured)
self.assertEqual(app.conf.worker_agent, 'foo:Bar')
self.assertEqual(app.conf.broker_url, 'foo://bar')
self.assertEqual(app._preconf['worker_agent'], 'foo:Bar')
def test_pending_configuration__compat_settings(self):
with self.Celery(broker='foo://bar', backend='foo') as app:
app.conf.update(
CELERY_ALWAYS_EAGER=4,
CELERY_DEFAULT_DELIVERY_MODE=63,
CELERYD_AGENT='foo:Barz',
)
self.assertEqual(app.conf.task_always_eager, 4)
self.assertEqual(app.conf.task_default_delivery_mode, 63)
self.assertEqual(app.conf.worker_agent, 'foo:Barz')
self.assertEqual(app.conf.broker_url, 'foo://bar')
self.assertEqual(app.conf.result_backend, 'foo')
def test_pending_configuration__setdefault(self):
with self.Celery(broker='foo://bar') as app:
app.conf.setdefault('worker_agent', 'foo:Bar')
self.assertFalse(app.configured)
def test_pending_configuration__iter(self):
with self.Celery(broker='foo://bar') as app:
app.conf.worker_agent = 'foo:Bar'
self.assertFalse(app.configured)
self.assertTrue(list(keys(app.conf)))
self.assertFalse(app.configured)
self.assertIn('worker_agent', app.conf)
self.assertFalse(app.configured)
self.assertTrue(dict(app.conf))
self.assertTrue(app.configured)
def test_pending_configuration__raises_ImproperlyConfigured(self):
with self.Celery() as app:
app.conf.worker_agent = 'foo://bar'
app.conf.task_default_delivery_mode = 44
app.conf.CELERY_ALWAYS_EAGER = True
with self.assertRaises(ImproperlyConfigured):
app.finalize()
def test_repr(self):
self.assertTrue(repr(self.app))
@@ -236,7 +302,7 @@ def test_custom_task_registry(self):
def test_include_argument(self):
with self.Celery(include=('foo', 'bar.foo')) as app:
self.assertEqual(app.conf.CELERY_IMPORTS, ('foo', 'bar.foo'))
self.assertEqual(app.conf.include, ('foo', 'bar.foo'))
def test_set_as_current(self):
current = _state._tls.current_app
@@ -324,7 +390,7 @@ def _inner(*args, **kwargs):
return fun(*args, **kwargs)
return _inner
self.app.conf.CELERY_ANNOTATIONS = {
self.app.conf.task_annotations = {
adX.name: {'@__call__': deco}
}
adX.bind(self.app)
@@ -416,7 +482,7 @@ def assert_config2(self):
def test_config_from_object__lazy(self):
conf = ObjectConfig2()
self.app.config_from_object(conf)
self.assertFalse(self.app.loader._conf)
self.assertIs(self.app.loader._conf, unconfigured)
self.assertIs(self.app._config_source, conf)
self.assert_config2()
@@ -427,46 +493,110 @@ def test_config_from_object__force(self):
self.assert_config2()
def test_config_from_object__compat(self):
class Config(object):
CELERY_ALWAYS_EAGER = 44
CELERY_DEFAULT_DELIVERY_MODE = 30
CELERY_TASK_PUBLISH_RETRY = False
self.app.config_from_object(Config)
self.assertEqual(self.app.conf.task_always_eager, 44)
self.assertEqual(self.app.conf.CELERY_ALWAYS_EAGER, 44)
self.assertFalse(self.app.conf.task_publish_retry)
self.assertEqual(self.app.conf.task_default_routing_key, 'celery')
def test_config_from_object__supports_old_names(self):
class Config(object):
task_always_eager = 44
task_default_delivery_mode = 301
self.app.config_from_object(Config())
self.assertEqual(self.app.conf.CELERY_ALWAYS_EAGER, 44)
self.assertEqual(self.app.conf.task_always_eager, 44)
self.assertEqual(self.app.conf.CELERY_DEFAULT_DELIVERY_MODE, 301)
self.assertEqual(self.app.conf.task_default_delivery_mode, 301)
self.assertEqual(self.app.conf.task_default_routing_key, 'testcelery')
def test_config_from_object__namespace_uppercase(self):
class Config(object):
CELERY_TASK_ALWAYS_EAGER = 44
CELERY_TASK_DEFAULT_DELIVERY_MODE = 301
self.app.config_from_object(Config(), namespace='CELERY_')
self.assertEqual(self.app.conf.task_always_eager, 44)
def test_config_from_object__namespace_lowercase(self):
class Config(object):
celery_task_always_eager = 44
celery_task_default_delivery_mode = 301
self.app.config_from_object(Config(), namespace='celery_')
self.assertEqual(self.app.conf.task_always_eager, 44)
def test_config_from_object__mixing_new_and_old(self):
class Config(object):
task_always_eager = 44
worker_agent = 'foo:Agent'
worker_consumer = 'foo:Consumer'
beat_schedule = '/foo/schedule'
CELERY_DEFAULT_DELIVERY_MODE = 301
with self.assertRaises(ImproperlyConfigured) as exc:
self.app.config_from_object(Config(), force=True)
self.assertTrue(
exc.args[0].startswith('CELERY_DEFAULT_DELIVERY_MODE'))
self.assertIn('task_default_delivery_mode', exc.args[0])
def test_config_from_object__mixing_old_and_new(self):
class Config(object):
CELERY_ALWAYS_EAGER = 44
CELERYD_AGENT = 'foo:Agent'
CELERYD_CONSUMER = 'foo:Consumer'
CELERYBEAT_SCHEDULE = '/foo/schedule'
task_default_delivery_mode = 301
with self.assertRaises(ImproperlyConfigured) as exc:
self.app.config_from_object(Config(), force=True)
self.assertTrue(
exc.args[0].startswith('task_default_delivery_mode'))
self.assertIn('CELERY_DEFAULT_DELIVERY_MODE', exc.args[0])
def test_config_from_cmdline(self):
cmdline = ['.always_eager=no',
'.result_backend=/dev/null',
'celeryd.prefetch_multiplier=368',
cmdline = ['task_always_eager=no',
'result_backend=/dev/null',
'worker_prefetch_multiplier=368',
'.foobarstring=(string)300',
'.foobarint=(int)300',
'.result_engine_options=(dict){"foo": "bar"}']
self.app.config_from_cmdline(cmdline, namespace='celery')
self.assertFalse(self.app.conf.CELERY_ALWAYS_EAGER)
self.assertEqual(self.app.conf.CELERY_RESULT_BACKEND, '/dev/null')
self.assertEqual(self.app.conf.CELERYD_PREFETCH_MULTIPLIER, 368)
self.assertEqual(self.app.conf.CELERY_FOOBARSTRING, '300')
self.assertEqual(self.app.conf.CELERY_FOOBARINT, 300)
self.assertDictEqual(self.app.conf.CELERY_RESULT_ENGINE_OPTIONS,
'sqlalchemy_engine_options=(dict){"foo": "bar"}']
self.app.config_from_cmdline(cmdline, namespace='worker')
self.assertFalse(self.app.conf.task_always_eager)
self.assertEqual(self.app.conf.result_backend, '/dev/null')
self.assertEqual(self.app.conf.worker_prefetch_multiplier, 368)
self.assertEqual(self.app.conf.worker_foobarstring, '300')
self.assertEqual(self.app.conf.worker_foobarint, 300)
self.assertDictEqual(self.app.conf.sqlalchemy_engine_options,
{'foo': 'bar'})
def test_compat_setting_CELERY_BACKEND(self):
self.app._preconf = {}
self.app.conf.defaults[0]['CELERY_RESULT_BACKEND'] = None
self.app.config_from_object(Object(CELERY_BACKEND='set_by_us'))
self.assertEqual(self.app.conf.CELERY_RESULT_BACKEND, 'set_by_us')
def test_setting_BROKER_TRANSPORT_OPTIONS(self):
def test_setting__broker_transport_options(self):
_args = {'foo': 'bar', 'spam': 'baz'}
self.app.config_from_object(Object())
self.assertEqual(self.app.conf.BROKER_TRANSPORT_OPTIONS, {})
self.assertEqual(self.app.conf.broker_transport_options, {})
self.app.config_from_object(Object(BROKER_TRANSPORT_OPTIONS=_args))
self.assertEqual(self.app.conf.BROKER_TRANSPORT_OPTIONS, _args)
self.app.config_from_object(Object(broker_transport_options=_args))
self.assertEqual(self.app.conf.broker_transport_options, _args)
def test_Windows_log_color_disabled(self):
self.app.IS_WINDOWS = True
self.assertFalse(self.app.log.supports_color(True))
def test_compat_setting_CARROT_BACKEND(self):
self.app.config_from_object(Object(CARROT_BACKEND='set_by_us'))
self.assertEqual(self.app.conf.BROKER_TRANSPORT, 'set_by_us')
def test_WorkController(self):
x = self.app.WorkController
self.assertIs(x.app, self.app)
@@ -537,9 +667,9 @@ def mail_admins(*args, **kwargs):
return args, kwargs
self.app.loader = Loader(app=self.app)
self.app.conf.ADMINS = None
self.app.conf.admins = None
self.assertFalse(self.app.mail_admins('Subject', 'Body'))
self.app.conf.ADMINS = [('George Costanza', 'george@vandelay.com')]
self.app.conf.admins = [('George Costanza', 'george@vandelay.com')]
self.assertTrue(self.app.mail_admins('Subject', 'Body'))
def test_amqp_get_broker_info(self):
@@ -550,8 +680,8 @@ def test_amqp_get_broker_info(self):
'virtual_host': '/'},
self.app.connection('pyamqp://').info(),
)
self.app.conf.BROKER_PORT = 1978
self.app.conf.BROKER_VHOST = 'foo'
self.app.conf.broker_port = 1978
self.app.conf.broker_vhost = 'foo'
self.assertDictContainsSubset(
{'port': 1978, 'virtual_host': 'foo'},
self.app.connection('pyamqp://:1978/foo').info(),
@@ -563,14 +693,14 @@ def test_amqp_get_broker_info(self):
def test_amqp_failover_strategy_selection(self):
# Test passing in a string and make sure the string
# gets there untouched
self.app.conf.BROKER_FAILOVER_STRATEGY = 'foo-bar'
self.app.conf.broker_failover_strategy = 'foo-bar'
self.assertEqual(
self.app.connection('amqp:////value').failover_strategy,
'foo-bar',
)
# Try passing in None
self.app.conf.BROKER_FAILOVER_STRATEGY = None
self.app.conf.broker_failover_strategy = None
self.assertEqual(
self.app.connection('amqp:////value').failover_strategy,
itertools.cycle,
@@ -580,16 +710,12 @@ def test_amqp_failover_strategy_selection(self):
def my_failover_strategy(it):
yield True
self.app.conf.BROKER_FAILOVER_STRATEGY = my_failover_strategy
self.app.conf.broker_failover_strategy = my_failover_strategy
self.assertEqual(
self.app.connection('amqp:////value').failover_strategy,
my_failover_strategy,
)
def test_BROKER_BACKEND_alias(self):
self.assertEqual(self.app.conf.BROKER_BACKEND,
self.app.conf.BROKER_TRANSPORT)
def test_after_fork(self):
p = self.app._pool = Mock()
self.app._after_fork(self.app)
View
16 celery/tests/app/test_beat.py
@@ -182,7 +182,7 @@ def not_sync():
self.assertFalse(s._do_sync.called)
def test_should_sync_increments_sync_every_counter(self):
self.app.conf.CELERYBEAT_SYNC_EVERY = 2
self.app.conf.beat_sync_every = 2
@self.app.task(shared=False)
def not_sync():
@@ -198,10 +198,10 @@ def not_sync():
s.apply_async(s.Entry(task=not_sync.name, app=self.app))
s._do_sync.assert_called_with()
self.app.conf.CELERYBEAT_SYNC_EVERY = 0
self.app.conf.beat_sync_every = 0
def test_sync_task_counter_resets_on_do_sync(self):
self.app.conf.CELERYBEAT_SYNC_EVERY = 1
self.app.conf.beat_sync_every = 1
@self.app.task(shared=False)
def not_sync():
@@ -214,7 +214,7 @@ def not_sync():
s.apply_async(s.Entry(task=not_sync.name, app=self.app))
self.assertEqual(s._tasks_since_sync, 0)
self.app.conf.CELERYBEAT_SYNC_EVERY = 0
self.app.conf.beat_sync_every = 0
@patch('celery.app.base.Celery.send_task')
def test_send_task(self, send_task):
@@ -249,20 +249,20 @@ def test_ensure_connection_error_handler(self, ensure):
callback(KeyError(), 5)
def test_install_default_entries(self):
self.app.conf.CELERY_TASK_RESULT_EXPIRES = None
self.app.conf.CELERYBEAT_SCHEDULE = {}
self.app.conf.result_expires = None
self.app.conf.beat_schedule = {}
s = mScheduler(app=self.app)
s.install_default_entries({})
self.assertNotIn('celery.backend_cleanup', s.data)
self.app.backend.supports_autoexpire = False
self.app.conf.CELERY_TASK_RESULT_EXPIRES = 30
self.app.conf.result_expires = 30
s = mScheduler(app=self.app)
s.install_default_entries({})
self.assertIn('celery.backend_cleanup', s.data)
self.app.backend.supports_autoexpire = True
self.app.conf.CELERY_TASK_RESULT_EXPIRES = 31
self.app.conf.result_expires = 31
s = mScheduler(app=self.app)
s.install_default_entries({})
self.assertNotIn('celery.backend_cleanup', s.data)
View
4 celery/tests/app/test_builtins.py
@@ -83,7 +83,7 @@ def setup(self):
def test_apply_async_eager(self):
self.task.apply = Mock()
self.app.conf.CELERY_ALWAYS_EAGER = True
self.app.conf.task_always_eager = True
self.task.apply_async((1, 2, 3, 4, 5))
self.assertTrue(self.task.apply.called)
@@ -208,7 +208,7 @@ def test_forward_options(self):
self.assertEqual(resbody.options['chord'], 'some_chord_id')
def test_apply_eager(self):
self.app.conf.CELERY_ALWAYS_EAGER = True
self.app.conf.task_always_eager = True
x = chord([self.add.s(i, i) for i in range(10)], body=self.xsum.s())
r = x.apply_async()
self.assertEqual(r.get(), 90)
View
29 celery/tests/app/test_defaults.py
@@ -4,11 +4,13 @@
from importlib import import_module
from celery.app.defaults import NAMESPACES
from celery.tests.case import (
AppCase, pypy_version, sys_platform,
from celery.app.defaults import (
_OLD_DEFAULTS, _OLD_SETTING_KEYS, _TO_NEW_KEY, _TO_OLD_KEY,
DEFAULTS, NAMESPACES, SETTING_KEYS
)
from celery.five import values
from celery.tests.case import AppCase, pypy_version, sys_platform
class test_defaults(AppCase):
@@ -21,7 +23,7 @@ def teardown(self):
sys.modules['celery.app.defaults'] = self._prev
def test_option_repr(self):
self.assertTrue(repr(NAMESPACES['BROKER']['URL']))
self.assertTrue(repr(NAMESPACES['broker']['url']))
def test_any(self):
val = object()
@@ -37,6 +39,21 @@ def test_default_pool_pypy_15(self):
with pypy_version((1, 5, 0)):
self.assertEqual(self.defaults.DEFAULT_POOL, 'prefork')
def test_compat_indices(self):
self.assertFalse(any(key.isupper() for key in DEFAULTS))
self.assertFalse(any(key.islower() for key in _OLD_DEFAULTS))
self.assertFalse(any(key.isupper() for key in _TO_OLD_KEY))
self.assertFalse(any(key.islower() for key in _TO_NEW_KEY))
self.assertFalse(any(key.isupper() for key in SETTING_KEYS))
self.assertFalse(any(key.islower() for key in _OLD_SETTING_KEYS))
self.assertFalse(any(value.isupper() for value in values(_TO_NEW_KEY)))
self.assertFalse(any(value.islower() for value in values(_TO_OLD_KEY)))
for key in _TO_NEW_KEY:
self.assertIn(key, _OLD_SETTING_KEYS)
for key in _TO_OLD_KEY:
self.assertIn(key, SETTING_KEYS)
def test_default_pool_jython(self):
with sys_platform('java 1.6.51'):
self.assertEqual(self.defaults.DEFAULT_POOL, 'threads')
@@ -46,7 +63,7 @@ def test_find(self):
self.assertEqual(find('server_email')[2].default, 'celery@localhost')
self.assertEqual(find('default_queue')[2].default, 'celery')
self.assertEqual(find('celery_default_exchange')[2], 'celery')
self.assertEqual(find('task_default_exchange')[2], 'celery')
@property
def defaults(self):
View
15 celery/tests/app/test_loaders.py
@@ -22,7 +22,7 @@
class DummyLoader(base.BaseLoader):
def read_configuration(self):
return {'foo': 'bar', 'CELERY_IMPORTS': ('os', 'sys')}
return {'foo': 'bar', 'imports': ('os', 'sys')}
class test_loaders(AppCase):
@@ -65,10 +65,9 @@ def test_now(self):
self.assertTrue(self.loader.now(utc=False))
def test_read_configuration_no_env(self):
self.assertDictEqual(
self.assertIsNone(
base.BaseLoader(app=self.app).read_configuration(
'FOO_X_S_WE_WQ_Q_WE'),
{},
)
def test_autodiscovery(self):
@@ -101,7 +100,7 @@ def test_conf_property(self):
def test_import_default_modules(self):
def modnames(l):
return [m.__name__ for m in l]
self.app.conf.CELERY_IMPORTS = ('os', 'sys')
self.app.conf.imports = ('os', 'sys')
self.assertEqual(
sorted(modnames(self.loader.import_default_modules())),
sorted(modnames([os, sys])),
@@ -183,17 +182,17 @@ class ConfigModule(ModuleType):
configname = os.environ.get('CELERY_CONFIG_MODULE') or 'celeryconfig'
celeryconfig = ConfigModule(configname)
celeryconfig.CELERY_IMPORTS = ('os', 'sys')
celeryconfig.imports = ('os', 'sys')
prevconfig = sys.modules.get(configname)
sys.modules[configname] = celeryconfig
try:
l = default.Loader(app=self.app)
l.find_module = Mock(name='find_module')
settings = l.read_configuration(fail_silently=False)
self.assertTupleEqual(settings.CELERY_IMPORTS, ('os', 'sys'))
self.assertTupleEqual(settings.imports, ('os', 'sys'))
settings = l.read_configuration(fail_silently=False)
self.assertTupleEqual(settings.CELERY_IMPORTS, ('os', 'sys'))
self.assertTupleEqual(settings.imports, ('os', 'sys'))
l.on_worker_init()
finally:
if prevconfig:
@@ -239,7 +238,7 @@ def setup(self):
self.loader = AppLoader(app=self.app)
def test_on_worker_init(self):
self.app.conf.CELERY_IMPORTS = ('subprocess',)
self.app.conf.imports = ('subprocess',)
sys.modules.pop('subprocess', None)
self.loader.init_worker()
self.assertIn('subprocess', sys.modules)
View
2 celery/tests/app/test_log.py
@@ -190,7 +190,7 @@ def test_setup_logging_subsystem_misc(self):
def test_setup_logging_subsystem_misc2(self):
with restore_logging():
self.app.conf.CELERYD_HIJACK_ROOT_LOGGER = True
self.app.conf.worker_hijack_root_logger = True
self.app.log.setup_logging_subsystem()
def test_get_default_logger(self):
View
14 celery/tests/app/test_routes.py
@@ -21,7 +21,7 @@ def expand(answer):
def set_queues(app, **queues):
app.conf.CELERY_QUEUES = queues
app.conf.task_queues = queues
app.amqp.queues = app.amqp.Queues(queues)
@@ -39,9 +39,9 @@ def setup(self):
'routing_key': 'b.b.#',
}
self.d_queue = {
'exchange': self.app.conf.CELERY_DEFAULT_EXCHANGE,
'exchange_type': self.app.conf.CELERY_DEFAULT_EXCHANGE_TYPE,
'routing_key': self.app.conf.CELERY_DEFAULT_ROUTING_KEY,
'exchange': self.app.conf.task_default_exchange,
'exchange_type': self.app.conf.task_default_exchange_type,
'routing_key': self.app.conf.task_default_routing_key,
}
@self.app.task(shared=False)
@@ -74,7 +74,7 @@ def test_route_for_task(self):
def test_expand_route_not_found(self):
expand = E(self.app, self.app.amqp.Queues(
self.app.conf.CELERY_QUEUES, False))
self.app.conf.task_queues, False))
route = routes.MapRoute({'a': {'queue': 'x'}})
with self.assertRaises(QueueNotFound):
expand(route.route_for_task('a'))
@@ -124,7 +124,7 @@ def test_expand_destination_string(self):
def test_lookup_paths_traversed(self):
set_queues(
self.app, foo=self.a_queue, bar=self.b_queue,
**{self.app.conf.CELERY_DEFAULT_QUEUE: self.d_queue}
**{self.app.conf.task_default_queue: self.d_queue}
)
R = routes.prepare((
{'celery.xaza': {'queue': 'bar'}},
@@ -135,7 +135,7 @@ def test_lookup_paths_traversed(self):
args=[1, 2], kwargs={})['queue'].name, 'foo')
self.assertEqual(
router.route({}, 'celery.poza')['queue'].name,
self.app.conf.CELERY_DEFAULT_QUEUE,
self.app.conf.task_default_queue,
)
View
2 celery/tests/backends/test_amqp.py
@@ -375,7 +375,7 @@ def se(*args, **kwargs):
def test_no_expires(self):
b = self.create_backend(expires=None)
app = self.app
app.conf.CELERY_TASK_RESULT_EXPIRES = None
app.conf.result_expires = None
b = self.create_backend(expires=None)
with self.assertRaises(KeyError):
b.queue_arguments['x-expires']
View
2 celery/tests/backends/test_base.py
@@ -343,7 +343,7 @@ def test_chord_part_return_propagate_default(self):
self.assertFalse(self.b.expire.called)
deps.delete.assert_called_with()
deps.join_native.assert_called_with(
propagate=self.b.app.conf.CELERY_CHORD_PROPAGATES,
propagate=self.b.app.conf.chord_propagates,
timeout=3.0,
)
View
4 celery/tests/backends/test_cache.py
@@ -31,12 +31,12 @@ def __init__(self, data):
class test_CacheBackend(AppCase):
def setup(self):
self.app.conf.CELERY_RESULT_SERIALIZER = 'pickle'
self.app.conf.result_serializer = 'pickle'
self.tb = CacheBackend(backend='memory://', app=self.app)
self.tid = uuid()
def test_no_backend(self):
self.app.conf.CELERY_CACHE_BACKEND = None
self.app.conf.cache_backend = None
with self.assertRaises(ImproperlyConfigured):
CacheBackend(backend=None, app=self.app)
View
12 celery/tests/backends/test_cassandra.py
@@ -48,9 +48,9 @@ class test_CassandraBackend(AppCase):
def setup(self):
self.app.conf.update(
CASSANDRA_SERVERS=['example.com'],
CASSANDRA_KEYSPACE='keyspace',
CASSANDRA_COLUMN_FAMILY='columns',
cassandra_servers=['example.com'],
cassandra_keyspace='keyspace',
cassandra_column_family='columns',
)
def test_init_no_pycassa(self):
@@ -71,16 +71,16 @@ def test_init_with_and_without_LOCAL_QUROM(self):
cons = mod.pycassa.ConsistencyLevel = Object()
cons.LOCAL_QUORUM = 'foo'
self.app.conf.CASSANDRA_READ_CONSISTENCY = 'LOCAL_FOO'
self.app.conf.CASSANDRA_WRITE_CONSISTENCY = 'LOCAL_FOO'
self.app.conf.cassandra_read_consistency = 'LOCAL_FOO'
self.app.conf.cassandra_write_consistency = 'LOCAL_FOO'
mod.CassandraBackend(app=self.app)
cons.LOCAL_FOO = 'bar'
mod.CassandraBackend(app=self.app)
# no servers raises ImproperlyConfigured
with self.assertRaises(ImproperlyConfigured):
self.app.conf.CASSANDRA_SERVERS = None
self.app.conf.cassandra_servers = None
mod.CassandraBackend(
app=self.app, keyspace='b', column_family='c',
)
View
14 celery/tests/backends/test_couchbase.py
@@ -47,13 +47,13 @@ def test_init_no_couchbase(self):
def test_init_no_settings(self):
"""Test init no settings."""
self.app.conf.CELERY_COUCHBASE_BACKEND_SETTINGS = []
self.app.conf.couchbase_backend_settings = []
with self.assertRaises(ImproperlyConfigured):
CouchBaseBackend(app=self.app)
def test_init_settings_is_None(self):
"""Test init settings is None."""
self.app.conf.CELERY_COUCHBASE_BACKEND_SETTINGS = None
self.app.conf.couchbase_backend_settings = None
CouchBaseBackend(app=self.app)
def test_get_connection_connection_exists(self):
@@ -75,7 +75,7 @@ def test_get(self):
TODO Should test on key not exists
"""
self.app.conf.CELERY_COUCHBASE_BACKEND_SETTINGS = {}
self.app.conf.couchbase_backend_settings = {}
x = CouchBaseBackend(app=self.app)
x._connection = Mock()
mocked_get = x._connection.get = Mock()
@@ -91,7 +91,7 @@ def test_set(self):
CouchBaseBackend.set should return None and take two params
db conn to couchbase is mocked.
"""
self.app.conf.CELERY_COUCHBASE_BACKEND_SETTINGS = None
self.app.conf.couchbase_backend_settings = None
x = CouchBaseBackend(app=self.app)
x._connection = MagicMock()
x._connection.set = MagicMock()
@@ -107,7 +107,7 @@ def test_delete(self):
TODO Should test on key not exists.
"""
self.app.conf.CELERY_COUCHBASE_BACKEND_SETTINGS = {}
self.app.conf.couchbase_backend_settings = {}
x = CouchBaseBackend(app=self.app)
x._connection = Mock()
mocked_delete = x._connection.delete = Mock()
@@ -120,9 +120,9 @@ def test_config_params(self):
"""
Test config params are correct.
celery.conf.CELERY_COUCHBASE_BACKEND_SETTINGS is properly set.
app.conf.couchbase_backend_settings is properly set.
"""
self.app.conf.CELERY_COUCHBASE_BACKEND_SETTINGS = {
self.app.conf.couchbase_backend_settings = {
'bucket': 'mycoolbucket',
'host': ['here.host.com', 'there.host.com'],
'username': 'johndoe',
View
4 celery/tests/backends/test_database.py
@@ -39,7 +39,7 @@ def setup(self):
if DatabaseBackend is None:
raise SkipTest('sqlalchemy not installed')
self.uri = 'sqlite:///test.db'
self.app.conf.CELERY_RESULT_SERIALIZER = 'pickle'
self.app.conf.result_serializer = 'pickle'
def test_retry_helper(self):
from celery.backends.database import DatabaseError
@@ -56,7 +56,7 @@ def raises():
self.assertEqual(calls[0], 5)
def test_missing_dburi_raises_ImproperlyConfigured(self):
self.app.conf.CELERY_RESULT_DBURI = None
self.app.conf.sqlalchemy_dburi = None
with self.assertRaises(ImproperlyConfigured):
DatabaseBackend(app=self.app)
View
8 celery/tests/backends/test_mongodb.py
@@ -59,16 +59,16 @@ def test_init_no_mongodb(self):
module.pymongo = prev
def test_init_no_settings(self):
self.app.conf.CELERY_MONGODB_BACKEND_SETTINGS = []
self.app.conf.mongodb_backend_settings = []
with self.assertRaises(ImproperlyConfigured):
MongoBackend(app=self.app)
def test_init_settings_is_None(self):
self.app.conf.CELERY_MONGODB_BACKEND_SETTINGS = None
self.app.conf.mongodb_backend_settings = None
MongoBackend(app=self.app)
def test_init_with_settings(self):
self.app.conf.CELERY_MONGODB_BACKEND_SETTINGS = None
self.app.conf.mongodb_backend_settings = None
# empty settings
mb = MongoBackend(app=self.app)
@@ -103,7 +103,7 @@ def test_init_with_settings(self):
self.assertEqual(mb.database_name, 'celerydatabase')
# same uri, change some parameters in backend settings
self.app.conf.CELERY_MONGODB_BACKEND_SETTINGS = {
self.app.conf.mongodb_backend_settings = {
'replicaset': 'rs1',
'user': 'backenduser',
'database': 'another_db',
View
12 celery/tests/backends/test_new_cassandra.py
@@ -20,9 +20,9 @@ class test_CassandraBackend(AppCase):
def setup(self):
self.app.conf.update(
CASSANDRA_SERVERS=['example.com'],
CASSANDRA_KEYSPACE='celery',
CASSANDRA_TABLE='task_results',
cassandra_servers=['example.com'],
cassandra_keyspace='celery',
cassandra_table='task_results',
)
def test_init_no_cassandra(self):
@@ -44,16 +44,16 @@ def test_init_with_and_without_LOCAL_QUROM(self):
cons = mod.cassandra.ConsistencyLevel = Object()
cons.LOCAL_QUORUM = 'foo'
self.app.conf.CASSANDRA_READ_CONSISTENCY = 'LOCAL_FOO'
self.app.conf.CASSANDRA_WRITE_CONSISTENCY = 'LOCAL_FOO'
self.app.conf.cassandra_read_consistency = 'LOCAL_FOO'
self.app.conf.cassandra_write_consistency = 'LOCAL_FOO'
mod.CassandraBackend(app=self.app)
cons.LOCAL_FOO = 'bar'
mod.CassandraBackend(app=self.app)
# no servers raises ImproperlyConfigured
with self.assertRaises(ImproperlyConfigured):
self.app.conf.CASSANDRA_SERVERS = None
self.app.conf.cassandra_servers = None
mod.CassandraBackend(
app=self.app, keyspace='b', column_family='c',
)
View
12 celery/tests/backends/test_redis.py
@@ -180,15 +180,15 @@ def test_compat_propertie(self):
def test_conf_raises_KeyError(self):
self.app.conf = AttributeDict({
'CELERY_RESULT_SERIALIZER': 'json',
'CELERY_MAX_CACHED_RESULTS': 1,
'CELERY_ACCEPT_CONTENT': ['json'],
'CELERY_TASK_RESULT_EXPIRES': None,
'result_serializer': 'json',
'result_cache_max': 1,
'result_expires': None,
'accept_content': ['json'],
})
self.Backend(app=self.app, new_join=True)
def test_expires_defaults_to_config(self):
self.app.conf.CELERY_TASK_RESULT_EXPIRES = 10
self.app.conf.result_expires = 10
b = self.Backend(expires=None, app=self.app, new_join=True)
self.assertEqual(b.expires, 10)
@@ -216,7 +216,7 @@ def test_expires_is_None(self):
b = self.Backend(expires=None, app=self.app, new_join=True)
self.assertEqual(
b.expires,
self.app.conf.CELERY_TASK_RESULT_EXPIRES.total_seconds(),
self.app.conf.result_expires.total_seconds(),
)
def test_expires_is_timedelta(self):
View
26 celery/tests/backends/test_riak.py
@@ -18,7 +18,7 @@ class test_RiakBackend(AppCase):
def setup(self):
if riak is None:
raise SkipTest('riak is not installed.')
self.app.conf.CELERY_RESULT_BACKEND = 'riak://'
self.app.conf.result_backend = 'riak://'
@property
def backend(self):
@@ -37,15 +37,15 @@ def test_init_no_riak(self):
def test_init_no_settings(self):
"""Test init no settings."""
self.app.conf.CELERY_RIAK_BACKEND_SETTINGS = []
self.app.conf.riak_backend_settings = []
with self.assertRaises(ImproperlyConfigured):
RiakBackend(app=self.app)
def test_init_settings_is_None(self):
"""
Test init settings is None
"""
self.app.conf.CELERY_RIAK_BACKEND_SETTINGS = None
self.app.conf.riak_backend_settings = None
self.assertTrue(self.app.backend)
def test_get_client_client_exists(self):
@@ -67,7 +67,7 @@ def test_get(self):
db conn to riak is mocked
TODO Should test on key not exists
"""
self.app.conf.CELERY_COUCHBASE_BACKEND_SETTINGS = {}
self.app.conf.couchbase_backend_settings = {}
self.backend._client = Mock(name='_client')
self.backend._bucket = Mock(name='_bucket')
mocked_get = self.backend._bucket.get = Mock(name='bucket.get')
@@ -84,7 +84,7 @@ def test_set(self):
db conn to couchbase is mocked.
"""
self.app.conf.CELERY_COUCHBASE_BACKEND_SETTINGS = None
self.app.conf.couchbase_backend_settings = None
self.backend._client = MagicMock()
self.backend._bucket = MagicMock()
self.backend._bucket.set = MagicMock()
@@ -100,7 +100,7 @@ def test_delete(self):
TODO Should test on key not exists
"""
self.app.conf.CELERY_COUCHBASE_BACKEND_SETTINGS = {}
self.app.conf.couchbase_backend_settings = {}
self.backend._client = Mock(name='_client')
self.backend._bucket = Mock(name='_bucket')
@@ -112,11 +112,11 @@ def test_delete(self):
def test_config_params(self):
"""
test celery.conf.CELERY_RIAK_BACKEND_SETTINGS
celery.conf.CELERY_RIAK_BACKEND_SETTINGS
test celery.conf.riak_backend_settingS
celery.conf.riak_backend_settingS
is properly set
"""
self.app.conf.CELERY_RIAK_BACKEND_SETTINGS = {
self.app.conf.riak_backend_settings = {
'bucket': 'mycoolbucket',
'host': 'there.host.com',
'port': '1234',
@@ -139,17 +139,17 @@ def test_backend_params_by_url(self):
"""
test get backend params by url
"""
self.app.conf.CELERY_RESULT_BACKEND = 'riak://myhost:123/mycoolbucket'
self.app.conf.result_backend = 'riak://myhost:123/mycoolbucket'
self.assertEqual(self.backend.bucket_name, 'mycoolbucket')
self.assertEqual(self.backend.host, 'myhost')
self.assertEqual(self.backend.port, 123)
def test_non_ASCII_bucket_raises(self):
"""test celery.conf.CELERY_RIAK_BACKEND_SETTINGS and
celery.conf.CELERY_RIAK_BACKEND_SETTINGS
"""test app.conf.riak_backend_settings and
app.conf.riak_backend_settings
is properly set
"""
self.app.conf.CELERY_RIAK_BACKEND_SETTINGS = {
self.app.conf.riak_backend_settings = {
'bucket': 'héhé',
'host': 'there.host.com',
'port': '1234',
View
6 celery/tests/bin/test_base.py
@@ -281,14 +281,14 @@ def test_say_chat_no_body(self):
def test_with_cmdline_config(self):
cmd = MockCommand(app=self.app)
cmd.enable_config_from_cmdline = True
cmd.namespace = 'celeryd'
cmd.namespace = 'worker'
rest = cmd.setup_app_from_commandline(argv=[
'--loglevel=INFO', '--',
'broker.url=amqp://broker.example.com',
'.prefetch_multiplier=100'])
self.assertEqual(cmd.app.conf.BROKER_URL,
self.assertEqual(cmd.app.conf.broker_url,
'amqp://broker.example.com')
self.assertEqual(cmd.app.conf.CELERYD_PREFETCH_MULTIPLIER, 100)
self.assertEqual(cmd.app.conf.worker_prefetch_multiplier, 100)
self.assertListEqual(rest, ['--loglevel=INFO'])
def test_find_app(self):
View
8 celery/tests/bin/test_worker.py
@@ -237,12 +237,12 @@ def test_init_queues(self):
self.assertIn('celery', app.amqp.queues)
self.assertNotIn('celery', app.amqp.queues.consume_from)
c.CELERY_CREATE_MISSING_QUEUES = False
c.task_create_missing_queues = False
del(app.amqp.queues)
with self.assertRaises(ImproperlyConfigured):
self.Worker(app=self.app).setup_queues(['image'])
del(app.amqp.queues)
c.CELERY_CREATE_MISSING_QUEUES = True
c.task_create_missing_queues = True
worker = self.Worker(app=self.app)
worker.setup_queues(['image'])
self.assertIn('image', app.amqp.queues.consume_from)
@@ -283,7 +283,7 @@ def test_warns_if_running_as_privileged_user(self, _exit):
with patch('os.getuid') as getuid:
getuid.return_value = 0
self.app.conf.CELERY_ACCEPT_CONTENT = ['pickle']
self.app.conf.accept_content = ['pickle']
worker = self.Worker(app=self.app)
worker.on_start()
_exit.assert_called_with(1)
@@ -297,7 +297,7 @@ def test_warns_if_running_as_privileged_user(self, _exit):
worker.on_start()
finally:
platforms.C_FORCE_ROOT = False
self.app.conf.CELERY_ACCEPT_CONTENT = ['json']
self.app.conf.accept_content = ['json']
with self.assertWarnsRegex(
RuntimeWarning,
r'absolutely not recommended'):
View
22 celery/tests/case.py
@@ -85,21 +85,21 @@
CELERY_TEST_CONFIG = {
#: Don't want log output when running suite.
'CELERYD_HIJACK_ROOT_LOGGER': False,
'CELERY_SEND_TASK_ERROR_EMAILS': False,
'CELERY_DEFAULT_QUEUE': 'testcelery',
'CELERY_DEFAULT_EXCHANGE': 'testcelery',
'CELERY_DEFAULT_ROUTING_KEY': 'testcelery',
'CELERY_QUEUES': (
'worker_hijack_root_logger': False,
'worker_log_color': False,
'task_send_error_emails': False,
'task_default_queue': 'testcelery',
'task_default_exchange': 'testcelery',
'task_default_routing_key': 'testcelery',
'task_queues': (
Queue('testcelery', routing_key='testcelery'),
),
'CELERY_ACCEPT_CONTENT': ('json', 'pickle'),
'CELERY_ENABLE_UTC': True,
'CELERY_TIMEZONE': 'UTC',
'CELERYD_LOG_COLOR': False,
'accept_content': ('json', 'pickle'),
'enable_utc': True,
'timezone': 'UTC',
# Mongo results tests (only executed if installed and running)
'CELERY_MONGODB_BACKEND_SETTINGS': {
'mongodb_backend_settings': {
'host': os.environ.get('MONGO_HOST') or 'localhost',
'port': os.environ.get('MONGO_PORT') or 27017,
'database': os.environ.get('MONGO_DB') or 'celery_unittests',
View
4 celery/tests/compat_modules/test_http.py
@@ -142,15 +142,15 @@ def test_dispatch_POST(self):
class test_URL(AppCase):
def test_URL_get_async(self):
self.app.conf.CELERY_ALWAYS_EAGER = True
self.app.conf.task_always_eager = True
with mock_urlopen(success_response(100)):
d = http.URL(
'http://example.com/mul', app=self.app,
).get_async(x=10, y=10)
self.assertEqual(d.get(), 100)
def test_URL_post_async(self):
self.app.conf.CELERY_ALWAYS_EAGER = True
self.app.conf.task_always_eager = True
with mock_urlopen(success_response(100)):
d = http.URL(
'http://example.com/mul', app=self.app,
View
4 celery/tests/compat_modules/test_sets.py
@@ -170,10 +170,10 @@ def apply(self, *args, **kwargs):
[self.MockTask.subtask((i, i)) for i in (2, 4, 8)],
app=self.app,
)
app.conf.CELERY_ALWAYS_EAGER = True
app.conf.task_always_eager = True
ts.apply_async()
self.assertEqual(ts.applied, 1)
app.conf.CELERY_ALWAYS_EAGER = False
app.conf.task_always_eager = False
with patch('celery.task.sets.get_current_worker_task') as gwt:
parent = gwt.return_value = Mock()
View
2 celery/tests/events/test_events.py
@@ -125,7 +125,7 @@ def test_enabled_disable(self):
self.assertTrue(dispatcher.enabled)
self.assertTrue(dispatcher.producer.channel)
self.assertEqual(dispatcher.producer.serializer,
self.app.conf.CELERY_EVENT_SERIALIZER)
self.app.conf.event_serializer)
created_channel = dispatcher.producer.channel
dispatcher.disable()
View
6 celery/tests/security/test_security.py
@@ -57,7 +57,7 @@ def test_setup_security(self):
disabled = registry._disabled_content_types
self.assertEqual(0, len(disabled))
self.app.conf.CELERY_TASK_SERIALIZER = 'json'
self.app.conf.task_serializer = 'json'
self.app.setup_security()
self.assertIn('application/x-python-serialize', disabled)
disabled.clear()
@@ -75,7 +75,7 @@ def effect(*args):
finally:
calls[0] += 1
self.app.conf.CELERY_TASK_SERIALIZER = 'auth'
self.app.conf.task_serializer = 'auth'
with mock_open(side_effect=effect):
with patch('celery.security.registry') as registry:
store = Mock()
@@ -85,7 +85,7 @@ def effect(*args):
registry._set_default_serializer.assert_called_with('auth')
def test_security_conf(self):
self.app.conf.CELERY_TASK_SERIALIZER = 'auth'
self.app.conf.task_serializer = 'auth'
with self.assertRaises(ImproperlyConfigured):
self.app.setup_security()
View
4 celery/tests/tasks/test_canvas.py
@@ -198,7 +198,7 @@ def test_chunks(self):
x()
gr.apply_async.assert_called_with((), {}, route_name=self.add.name)
self.app.conf.CELERY_ALWAYS_EAGER = True
self.app.conf.task_always_eager = True
chunks.apply_chunks(app=self.app, **x['kwargs'])
@@ -216,7 +216,7 @@ def test_reverse(self):
self.assertIsInstance(signature(dict(x)), chain)
def test_always_eager(self):
self.app.conf.CELERY_ALWAYS_EAGER = True
self.app.conf.task_always_eager = True
self.assertEqual(~(self.add.s(4, 4) | self.add.s(8)), 16)
def test_apply(self):
View
6 celery/tests/tasks/test_chord.py
@@ -194,18 +194,18 @@ def addX(x, y):
def sumX(n):
return sum(n)
self.app.conf.CELERY_ALWAYS_EAGER = True
self.app.conf.task_always_eager = True
x = chord(addX.s(i, i) for i in range(10))
body = sumX.s()
result = x(body)
self.assertEqual(result.get(), sum(i + i for i in range(10)))
def test_apply(self):
self.app.conf.CELERY_ALWAYS_EAGER = False
self.app.conf.task_always_eager = False
from celery import chord
m = Mock()
m.app.conf.CELERY_ALWAYS_EAGER = False
m.app.conf.task_always_eager = False
m.AsyncResult = AsyncResult
prev, chord.run = chord.run, m
try:
Oops, something went wrong.