Code now works on both Python 3 and Python 2 (without using 2to3)

celery/__init__.py

@@ -27,16 +27,16 @@
 import os
 if os.environ.get('C_IMPDEBUG'):
     import sys
-    import __builtin__
-    real_import = __builtin__.__import__
+    from .five import builtins
+    real_import = builtins.__import__
 
     def debug_import(name, locals=None, globals=None, fromlist=None,
             level=-1):
         glob = globals or getattr(sys, 'emarfteg_'[::-1])(1).f_globals
         importer_name = glob and glob.get('__name__') or 'unknown'
         print('-- {0} imports {1}'.format(importer_name, name))
         return real_import(name, locals, globals, fromlist, level)
-    __builtin__.__import__ = debug_import
+    builtins.__import__ = debug_import
 
 STATICA_HACK = True
 globals()['kcah_acitats'[::-1].upper()] = False
@@ -54,7 +54,7 @@ def debug_import(name, locals=None, globals=None, fromlist=None,
     from celery.utils import uuid                       # noqa
 
 # Lazy loading
-from .__compat__ import recreate_module
+from .five import recreate_module
 
 old_module, new_module = recreate_module(__name__,  # pragma: no cover
     by_module={

celery/app/__init__.py

@@ -10,6 +10,8 @@
 
 import os
 
+from collections import Callable
+
 from celery.local import Proxy
 from celery import _state
 from celery._state import (  # noqa
@@ -127,6 +129,6 @@ def task_by_cons():
             return Proxy(task_by_cons)
         return __inner
 
-    if len(args) == 1 and callable(args[0]):
+    if len(args) == 1 and isinstance(args[0], Callable):
         return create_shared_task(**kwargs)(args[0])
     return create_shared_task(**kwargs)

celery/app/abstract.py

@@ -9,6 +9,8 @@
 """
 from __future__ import absolute_import
 
+from celery.five import items, with_metaclass
+
 
 class from_config(object):
 
@@ -23,7 +25,7 @@ class _configurated(type):
 
     def __new__(cls, name, bases, attrs):
         attrs['__confopts__'] = dict((attr, spec.get_key(attr))
-                                          for attr, spec in attrs.iteritems()
+                                          for attr, spec in items(attrs)
                                               if isinstance(spec, from_config))
         inherit_from = attrs.get('inherit_confopts', ())
         for subcls in bases:
@@ -34,18 +36,18 @@ def __new__(cls, name, bases, attrs):
         for subcls in inherit_from:
             attrs['__confopts__'].update(subcls.__confopts__)
         attrs = dict((k, v if not isinstance(v, from_config) else None)
-                        for k, v in attrs.iteritems())
+                        for k, v in items(attrs))
         return super(_configurated, cls).__new__(cls, name, bases, attrs)
 
 
+@with_metaclass(_configurated)
 class configurated(object):
-    __metaclass__ = _configurated
 
     def setup_defaults(self, kwargs, namespace='celery'):
         confopts = self.__confopts__
         app, find = self.app, self.app.conf.find_value_for_key
 
-        for attr, keyname in confopts.iteritems():
+        for attr, keyname in items(confopts):
             try:
                 value = kwargs[attr]
             except KeyError:
@@ -55,7 +57,7 @@ def setup_defaults(self, kwargs, namespace='celery'):
                     value = find(keyname, namespace)
             setattr(self, attr, value)
 
-        for attr_name, attr_value in kwargs.iteritems():
+        for attr_name, attr_value in items(kwargs):
             if attr_name not in confopts and attr_value is not None:
                 setattr(self, attr_name, attr_value)
 

celery/app/amqp.py

@@ -18,6 +18,7 @@
 from kombu.utils.encoding import safe_repr
 
 from celery import signals
+from celery.five import items
 from celery.utils.text import indent as textindent
 
 from . import app_or_default
@@ -55,7 +56,7 @@ def __init__(self, queues=None, default_exchange=None,
         self.ha_policy = ha_policy
         if isinstance(queues, (tuple, list)):
             queues = dict((q.name, q) for q in queues)
-        for name, q in (queues or {}).iteritems():
+        for name, q in items(queues or {}):
             self.add(q) if isinstance(q, Queue) else self.add_compat(name, **q)
 
     def __getitem__(self, name):
@@ -119,7 +120,7 @@ def format(self, indent=0, indent_first=True):
         if not active:
             return ''
         info = [QUEUE_FORMAT.strip().format(q)
-                    for _, q in sorted(active.iteritems())]
+                    for _, q in sorted(items(active))]
         if indent_first:
             return textindent('\n'.join(info), indent)
         return info[0] + '\n' + textindent('\n'.join(info[1:]), indent)
@@ -270,7 +271,8 @@ class TaskConsumer(Consumer):
     def __init__(self, channel, queues=None, app=None, **kw):
         self.app = app or self.app
         super(TaskConsumer, self).__init__(channel,
-                queues or self.app.amqp.queues.consume_from.values(), **kw)
+                queues or list(self.app.amqp.queues.consume_from.values()),
+                **kw)
 
 
 class AMQP(object):

celery/app/annotations.py

@@ -12,6 +12,7 @@
 """
 from __future__ import absolute_import
 
+from celery.five import string_t
 from celery.utils.functional import firstmethod, mpromise
 from celery.utils.imports import instantiate
 
@@ -44,7 +45,7 @@ def prepare(annotations):
     def expand_annotation(annotation):
         if isinstance(annotation, dict):
             return MapAnnotation(annotation)
-        elif isinstance(annotation, basestring):
+        elif isinstance(annotation, string_t):
             return mpromise(instantiate, annotation)
         return annotation
 

celery/app/base.py

@@ -11,7 +11,7 @@
 import threading
 import warnings
 
-from collections import defaultdict, deque
+from collections import Callable, defaultdict, deque
 from contextlib import contextmanager
 from copy import deepcopy
 from functools import wraps
@@ -22,10 +22,11 @@
 from kombu.utils import cached_property
 
 from celery import platforms
+from celery._state import _task_stack, _tls, get_current_app, _register_app
 from celery.exceptions import AlwaysEagerIgnored
+from celery.five import items, values
 from celery.loaders import get_loader_cls
 from celery.local import PromiseProxy, maybe_evaluate
-from celery._state import _task_stack, _tls, get_current_app, _register_app
 from celery.utils.functional import first
 from celery.utils.imports import instantiate, symbol_by_name
 
@@ -153,7 +154,7 @@ def _create_task_cls(fun):
 
             return _create_task_cls
 
-        if len(args) == 1 and callable(args[0]):
+        if len(args) == 1 and isinstance(args[0], Callable):
             return inner_create_task_cls(**opts)(*args)
         return inner_create_task_cls(**opts)
 
@@ -180,11 +181,11 @@ def finalize(self):
                 while pending:
                     maybe_evaluate(pending.popleft())
 
-                for task in self._tasks.itervalues():
+                for task in values(self._tasks):
                     task.bind(self)
 
     def add_defaults(self, fun):
-        if not callable(fun):
+        if not isinstance(fun, Callable):
             d, fun = fun, lambda: d
         if self.configured:
             return self.conf.add_defaults(fun())
@@ -334,7 +335,7 @@ def _get_config(self):
         while pending:
             s.add_defaults(pending.popleft()())
         if self._preconf:
-            for key, value in self._preconf.iteritems():
+            for key, value in items(self._preconf):
                 setattr(s, key, value)
         return s
 

celery/app/builtins.py

@@ -10,7 +10,6 @@
 from __future__ import absolute_import
 
 from collections import deque
-from itertools import imap, izip, starmap
 
 from celery._state import get_current_worker_task
 from celery.utils import uuid
@@ -91,7 +90,7 @@ def add_map_task(app):
     @app.task(name='celery.map')
     def xmap(task, it):
         task = subtask(task).type
-        return list(imap(task, it))
+        return [task(item) for item in it]
     return xmap
 
 
@@ -102,7 +101,7 @@ def add_starmap_task(app):
     @app.task(name='celery.starmap')
     def xstarmap(task, it):
         task = subtask(task).type
-        return list(starmap(task, it))
+        return [task(*item) for item in it]
     return xstarmap
 
 
@@ -160,7 +159,7 @@ def prepare_member(task):
                 return task, AsyncResult(tid)
 
             try:
-                tasks, res = list(izip(*[prepare_member(task)
+                tasks, res = list(zip(*[prepare_member(task)
                                                 for task in tasks]))
             except ValueError:  # tasks empty
                 tasks, res = [], []
@@ -323,8 +322,8 @@ def apply_async(self, args=(), kwargs={}, task_id=None, **options):
                 opt_value = options.pop(opt_name, None)
                 if opt_value:
                     body.set(**{opt_name: opt_value})
-            map(body.link, options.pop('link', []))
-            map(body.link_error, options.pop('link_error', []))
+            [body.link(s) for s in options.pop('link', [])]
+            [body.link_error(s) for s in options.pop('link_error', [])]
             callback_id = body.options.setdefault('task_id', task_id or uuid())
             parent = super(Chord, self).apply_async((header, body, args),
                                                      kwargs, **options)

celery/app/defaults.py

@@ -13,6 +13,7 @@
 from collections import deque
 from datetime import timedelta
 
+from celery.five import items
 from celery.utils import strtobool
 from celery.utils.functional import memoize
 
@@ -51,7 +52,7 @@ class Option(object):
     def __init__(self, default=None, *args, **kwargs):
         self.default = default
         self.type = kwargs.get('type') or 'string'
-        for attr, value in kwargs.iteritems():
+        for attr, value in items(kwargs):
             setattr(self, attr, value)
 
     def to_python(self, value):
@@ -215,7 +216,7 @@ def flatten(d, ns=''):
     stack = deque([(ns, d)])
     while stack:
         name, space = stack.popleft()
-        for key, value in space.iteritems():
+        for key, value in items(space):
             if isinstance(value, dict):
                 stack.append((name + key + '_', value))
             else:
@@ -242,7 +243,7 @@ def find(name, namespace='celery'):
         return namespace, name.upper(), NAMESPACES[namespace][name.upper()]
     except KeyError:
         # - Try all the other namespaces.
-        for ns, keys in NAMESPACES.iteritems():
+        for ns, keys in items(NAMESPACES):
             if ns.upper() == name.upper():
                 return None, ns, keys
             elif isinstance(keys, dict):

celery/app/log.py

@@ -22,6 +22,7 @@
 
 from celery import signals
 from celery._state import get_current_task
+from celery.five import string_t
 from celery.utils import isatty
 from celery.utils.log import (
     get_logger, mlevel,
@@ -31,7 +32,7 @@
 )
 from celery.utils.term import colored
 
-is_py3k = sys.version_info[0] == 3
+PY3 = sys.version_info[0] == 3
 
 
 class TaskFormatter(ColorFormatter):
@@ -85,7 +86,7 @@ def setup_logging_subsystem(self, loglevel=None, logfile=None,
         format = format or self.format
         colorize = self.supports_color(colorize, logfile)
         reset_multiprocessing_logger()
-        if not is_py3k:
+        if not PY3:
             ensure_process_aware_logger()
         receivers = signals.setup_logging.send(sender=None,
                         loglevel=loglevel, logfile=logfile,
@@ -110,7 +111,7 @@ def setup_logging_subsystem(self, loglevel=None, logfile=None,
 
         # This is a hack for multiprocessing's fork+exec, so that
         # logging before Process.run works.
-        logfile_name = logfile if isinstance(logfile, basestring) else ''
+        logfile_name = logfile if isinstance(logfile, string_t) else ''
         os.environ.update(_MP_FORK_LOGLEVEL_=str(loglevel),
                           _MP_FORK_LOGFILE_=logfile_name,
                           _MP_FORK_LOGFORMAT_=format)

celery/app/registry.py

@@ -14,6 +14,7 @@
 
 from celery._state import get_current_app
 from celery.exceptions import NotRegistered
+from celery.five import items
 
 
 class TaskRegistry(dict):
@@ -54,7 +55,7 @@ def periodic(self):
         return self.filter_types('periodic')
 
     def filter_types(self, type):
-        return dict((name, task) for name, task in self.iteritems()
+        return dict((name, task) for name, task in items(self)
                                 if getattr(task, 'type', 'regular') == type)
 
 

celery/app/routes.py

@@ -10,6 +10,7 @@
 from __future__ import absolute_import
 
 from celery.exceptions import QueueNotFound
+from celery.five import string_t
 from celery.utils import lpmerge
 from celery.utils.functional import firstmethod, mpromise
 from celery.utils.imports import instantiate
@@ -51,7 +52,7 @@ def route(self, options, task, args=(), kwargs={}):
 
     def expand_destination(self, route):
         # Route can be a queue name: convenient for direct exchanges.
-        if isinstance(route, basestring):
+        if isinstance(route, string_t):
             queue, route = route, {}
         else:
             # can use defaults from configured queue, but override specific
@@ -84,7 +85,7 @@ def prepare(routes):
     def expand_route(route):
         if isinstance(route, dict):
             return MapRoute(route)
-        if isinstance(route, basestring):
+        if isinstance(route, string_t):
             return mpromise(instantiate, route)
         return route