Py3 fixes

celery/app/amqp.py

@@ -18,7 +18,7 @@ from kombu.utils import cached_property, uuid
 from kombu.utils.encoding import safe_repr
 
 from celery import signals
-from celery.five import items
+from celery.five import items, string_t
 from celery.utils.text import indent as textindent
 
 from . import app_or_default
@@ -195,7 +195,7 @@ class TaskProducer(Producer):
         if queue is None and exchange is None:
             queue = self.default_queue
         if queue is not None:
-            if isinstance(queue, basestring):
+            if isinstance(queue, string_t):
                 qname, queue = queue, self.queues[queue]
             else:
                 qname = queue.name

celery/app/builtins.py

@@ -81,8 +81,8 @@ def add_unlock_chord_task(app):
             callback = subtask(callback)
             try:
                 ret = j(propagate=propagate)
-            except Exception, exc:
-                culprit = deps._failed_join_report().next()
+            except Exception as exc:
+                culprit = next(deps._failed_join_report())
 
                 app._tasks[callback.task].backend.fail_from_current_stack(
                     callback.id, exc=ChordError('Dependency %s raised %r' % (

celery/backends/base.py

@@ -439,8 +439,8 @@ class KeyValueStoreBackend(BaseBackend):
             callback = subtask(task.request.chord)
             try:
                 ret = j(propagate=propagate)
-            except Exception, exc:
-                culprit = deps._failed_join_report().next()
+            except Exception as exc:
+                culprit = next(deps._failed_join_report())
                 self.app._tasks[callback.task].backend.fail_from_current_stack(
                     callback.id, exc=ChordError('Dependency %s raised %r' % (
                         culprit.id, exc))

celery/bin/graph.py

@@ -160,7 +160,7 @@ class graph(Command):
             threads = threads[0:3] + [threads[-1]]
         for i, threads in enumerate(threads):
             threads_for[workers[i]] = maybe_abbr(
-                range(int(threads)), 'P', Tmax,
+                list(range(int(threads))), 'P', Tmax,
             )
 
         broker = Broker(args.get('broker', self.app.connection().as_uri()))

celery/bin/multi.py

@@ -435,7 +435,7 @@ def multi_args(p, cmd='celery worker', append='', prefix='', suffix=''):
         except ValueError:
             pass
         else:
-            names = list(map(str, range(1, noderange + 1)))
+            names = [str(n) for n in range(1, noderange + 1)]
             prefix = 'celery'
     cmd = options.pop('--cmd', cmd)
     append = options.pop('--append', append)
@@ -545,8 +545,9 @@ def parse_ns_range(ns, ranges=False):
     for space in ',' in ns and ns.split(',') or [ns]:
         if ranges and '-' in space:
             start, stop = space.split('-')
-            x = list(map(str, range(int(start), int(stop) + 1)))
-            ret.extend(x)
+            ret.extend(
+                str(n) for n in range(int(start), int(stop) + 1)
+            )
         else:
             ret.append(space)
     return ret

celery/bootsteps.py

@@ -113,7 +113,7 @@ class Namespace(object):
         self.state = RUN
         if self.on_start:
             self.on_start()
-        for i, step in enumerate(filter(None, parent.steps)):
+        for i, step in enumerate(s for s in parent.steps if s is not None):
             self._debug('Starting %s', step.alias)
             self.started = i + 1
             step.start(parent)

celery/canvas.py

@@ -12,7 +12,7 @@
 from __future__ import absolute_import
 
 from copy import deepcopy
-from functools import partial as _partial
+from functools import partial as _partial, reduce
 from operator import itemgetter
 from itertools import chain as _chain
 

celery/events/__init__.py

@@ -173,7 +173,7 @@ class EventDispatcher(object):
                 return
             try:
                 self._send(type, fields, self.producer, blind)
-            except Exception, exc:
+            except Exception as exc:
                 if not self.buffer_while_offline:
                     raise
                 self._outbound_buffer.append((type, fields, exc))

celery/five.py

@@ -302,7 +302,7 @@ def create_module(name, attrs, cls_attrs=None, pkg=None,
     cls_attrs = {} if cls_attrs is None else cls_attrs
 
     attrs = dict((attr_name, prepare_attr(attr) if prepare_attr else attr)
-                 for attr_name, attr in attrs.items())
+                 for attr_name, attr in items(attrs))
     module = sys.modules[fqdn] = type(name, (base, ), cls_attrs)(fqdn)
     module.__dict__.update(attrs)
     return module
@@ -344,6 +344,6 @@ def get_compat_module(pkg, name):
 
 def get_origins(defs):
     origins = {}
-    for module, items in defs.items():
-        origins.update(dict((item, module) for item in items))
+    for module, attrs in items(defs):
+        origins.update(dict((attr, module) for attr in attrs))
     return origins

celery/platforms.py

@@ -20,7 +20,7 @@ from billiard import current_process
 from contextlib import contextmanager
 
 from .local import try_import
-from .five import items, map, reraise, string_t
+from .five import items, map, range, reraise, string_t
 
 _setproctitle = try_import('setproctitle')
 resource = try_import('resource')

celery/schedules.py

@@ -16,7 +16,7 @@ from datetime import datetime, timedelta
 from kombu.utils import cached_property
 
 from . import current_app
-from .five import string_t
+from .five import range, string_t
 from .utils import is_iterable
 from .utils.timeutils import (
     timedelta_seconds, weekday, maybe_timedelta, remaining,
@@ -208,10 +208,9 @@ class crontab_parser(object):
         if len(toks) > 1:
             to = self._expand_number(toks[1])
             if to < fr:  # Wrap around max_ if necessary
-                return range(fr,
-                             self.min_ + self.max_) + range(self.min_,
-                                                            to + 1)
-            return range(fr, to + 1)
+                return (list(range(fr, self.min_ + self.max_)) +
+                        list(range(self.min_, to + 1)))
+            return list(range(fr, to + 1))
         return [fr]
 
     def _range_steps(self, toks):
@@ -225,7 +224,7 @@ class crontab_parser(object):
         return self._expand_star()[::int(toks[0])]
 
     def _expand_star(self, *args):
-        return range(self.min_, self.max_ + self.min_)
+        return list(range(self.min_, self.max_ + self.min_))
 
     def _expand_number(self, s):
         if isinstance(s, string_t) and s[0] == '-':

celery/task/http.py

@@ -114,7 +114,7 @@ class MutableURL(object):
                       ';{0}'.format(params) if params else '',
                       '?{0}'.format(query) if query else '',
                       '#{0}'.format(fragment) if fragment else '']
-        return ''.join(filter(None, components))
+        return ''.join(c for c in components if c)
 
     def __repr__(self):
         return '<{0}: {1}>'.format(type(self).__name__, self)

celery/utils/__init__.py

@@ -12,7 +12,6 @@ import os
 import sys
 import traceback
 import warnings
-import types
 import datetime
 
 from functools import partial, wraps
@@ -201,9 +200,9 @@ def strtobool(term, table={'false': False, 'no': False, '0': False,
     return term
 
 
-def jsonify(obj):
+def jsonify(obj, builtin_types=(int, float, string_t)):
     """Transforms object making it suitable for json serialization"""
-    if isinstance(obj, (int, float, string_t, types.NoneType)):
+    if obj is None or isinstance(obj, builtin_types):
         return obj
     elif isinstance(obj, (tuple, list)):
         return [jsonify(o) for o in obj]
@@ -247,7 +246,7 @@ def gen_task_name(app, name, module_name):
             module_name = '__main__'
     if module_name == '__main__' and app.main:
         return '.'.join([app.main, name])
-    return '.'.join(filter(None, [module_name, name]))
+    return '.'.join(p for p in [module_name, name] if p)
 
 
 def nodename(name, hostname):