Don't use future_builtins, it's not available in py3k

celery/__compat__.py

@@ -15,8 +15,8 @@ import operator
 import sys
 
 from functools import reduce
-from future_builtins import map
 from importlib import import_module
+from itertools import imap
 from types import ModuleType
 
 from .local import Proxy
@@ -169,7 +169,7 @@ def recreate_module(name, compat_modules=(), by_module={}, direct={},
     cattrs = dict(_compat_modules=compat_modules,
                   _all_by_module=by_module, _direct=direct,
                   _object_origins=origins,
-                  __all__=tuple(set(reduce(operator.add, map(tuple, [
+                  __all__=tuple(set(reduce(operator.add, imap(tuple, [
                                 compat_modules, origins, direct, attrs])))))
     new_module = create_module(name, attrs, cls_attrs=cattrs, base=base)
     new_module.__dict__.update(dict((mod, get_compat_module(new_module, mod))

celery/app/annotations.py

@@ -12,8 +12,6 @@
 """
 from __future__ import absolute_import
 
-from future_builtins import filter
-
 from celery.utils.functional import firstmethod, mpromise
 from celery.utils.imports import instantiate
 
@@ -22,7 +20,7 @@ _first_match_any = firstmethod('annotate_any')
 
 
 def resolve_all(anno, task):
-    return filter(None, (_first_match(anno, task), _first_match_any(anno)))
+    return (x for x in (_first_match(anno, task), _first_match_any(anno)) if x)
 
 
 class MapAnnotation(dict):

celery/app/builtins.py

@@ -10,8 +10,7 @@
 from __future__ import absolute_import
 
 from collections import deque
-from future_builtins import map, zip
-from itertools import starmap
+from itertools import imap, izip, starmap
 
 from celery._state import get_current_worker_task
 from celery.utils import uuid
@@ -92,7 +91,7 @@ def add_map_task(app):
     @app.task(name='celery.map')
     def xmap(task, it):
         task = subtask(task).type
-        return list(map(task, it))
+        return list(imap(task, it))
     return xmap
 
 
@@ -161,7 +160,7 @@ def add_group_task(app):
                 return task, AsyncResult(tid)
 
             try:
-                tasks, res = list(zip(*[prepare_member(task)
+                tasks, res = list(izip(*[prepare_member(task)
                                                 for task in tasks]))
             except ValueError:  # tasks empty
                 tasks, res = [], []

celery/backends/base.py

@@ -17,7 +17,7 @@ import time
 import sys
 
 from datetime import timedelta
-from future_builtins import map
+from itertools import imap
 
 from kombu import serialization
 from kombu.utils.encoding import bytes_to_str, ensure_bytes, from_utf8
@@ -351,7 +351,7 @@ class KeyValueStoreBackend(BaseBackend):
             r = self._mget_to_results(self.mget([self.get_key_for_task(k)
                                                     for k in keys]), keys)
             self._cache.update(r)
-            ids ^= set(map(bytes_to_str, r))
+            ids ^= set(imap(bytes_to_str, r))
             for key, value in r.iteritems():
                 yield bytes_to_str(key), value
             if timeout and iterations * interval >= timeout:

celery/bin/base.py

@@ -65,7 +65,7 @@ import sys
 import warnings
 
 from collections import defaultdict
-from future_builtins import zip
+from itertools import izip
 from optparse import OptionParser, IndentedHelpFormatter, make_option as Option
 from types import ModuleType
 
@@ -328,7 +328,7 @@ class Command(object):
         opts = {}
         for opt in self.preload_options:
             for t in (opt._long_opts, opt._short_opts):
-                opts.update(dict(zip(t, [opt.dest] * len(t))))
+                opts.update(dict(izip(t, [opt.dest] * len(t))))
         index = 0
         length = len(args)
         while index < length:

celery/bin/celery.py

@@ -13,9 +13,8 @@ import heapq
 import sys
 import warnings
 
-from future_builtins import map
-
 from importlib import import_module
+from itertools import imap
 from pprint import pformat
 
 from celery.platforms import EX_OK, EX_FAILURE, EX_UNAVAILABLE, EX_USAGE
@@ -524,7 +523,7 @@ class _RemoteControl(Command):
         destination = kwargs.get('destination')
         timeout = kwargs.get('timeout') or self.choices[method][0]
         if destination and isinstance(destination, basestring):
-            destination = list(map(str.strip, destination.split(',')))
+            destination = list(imap(str.strip, destination.split(',')))
 
         try:
             handler = getattr(self, method)

celery/bin/celeryd_multi.py

@@ -97,7 +97,7 @@ import socket
 import sys
 
 from collections import defaultdict
-from future_builtins import map
+from itertools import imap
 from subprocess import Popen
 from time import sleep
 
@@ -427,7 +427,7 @@ def multi_args(p, cmd='celeryd', append='', prefix='', suffix=''):
         except ValueError:
             pass
         else:
-            names = list(map(str, range(1, noderange + 1)))
+            names = list(imap(str, range(1, noderange + 1)))
             prefix = 'celery'
     cmd = options.pop('--cmd', cmd)
     append = options.pop('--append', append)
@@ -528,19 +528,19 @@ 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)))
+            x = list(imap(str, range(int(start), int(stop) + 1)))
             ret.extend(x)
         else:
             ret.append(space)
     return ret
 
 
-def abbreviations(map):
+def abbreviations(mapping):
 
     def expand(S):
         ret = S
         if S is not None:
-            for short, long in map.items():
+            for short, long in mapping.items():
                 ret = ret.replace(short, long)
         return ret
 

celery/canvas.py

@@ -11,9 +11,8 @@
 """
 from __future__ import absolute_import
 
-from future_builtins import map
 from operator import itemgetter
-from itertools import chain as _chain
+from itertools import chain as _chain, imap
 
 from kombu.utils import fxrange, kwdict, reprcall
 
@@ -219,7 +218,7 @@ class chain(Signature):
         return chain(*d['kwargs']['tasks'], **kwdict(d['options']))
 
     def __repr__(self):
-        return ' | '.join(map(repr, self.tasks))
+        return ' | '.join(imap(repr, self.tasks))
 Signature.register_type(chain)
 
 

celery/contrib/rdb.py

@@ -41,7 +41,7 @@ import os
 import socket
 import sys
 
-from future_builtins import map
+from itertools import imap
 from pdb import Pdb
 
 from billiard import current_process
@@ -95,7 +95,7 @@ class Rdb(Pdb):
         self.say(BANNER.format(self=self))
 
         self._client, address = self._sock.accept()
-        self.remote_addr = ':'.join(map(str, address))
+        self.remote_addr = ':'.join(imap(str, address))
         self.say(SESSION_STARTED.format(self=self))
         self._handle = sys.stdin = sys.stdout = self._client.makefile('rw')
         Pdb.__init__(self, completekey='tab',

celery/loaders/base.py

@@ -13,8 +13,8 @@ import importlib
 import os
 import re
 
-from future_builtins import map
 from datetime import datetime
+from itertools import imap
 
 from kombu.utils.encoding import safe_str
 
@@ -191,7 +191,7 @@ class BaseLoader(object):
                     raise ValueError('{0!r}: {1}'.format(ns_key, exc))
             return ns_key, value
 
-        return dict(map(getarg, args))
+        return dict(imap(getarg, args))
 
     def mail_admins(self, subject, body, fail_silently=False,
             sender=None, to=None, host=None, port=None,

celery/platforms.py

@@ -18,7 +18,7 @@ import signal as _signal
 import sys
 
 from contextlib import contextmanager
-from future_builtins import map
+from itertools import imap
 
 from .local import try_import
 
@@ -58,9 +58,9 @@ def pyimplementation():
     elif sys.platform.startswith('java'):
         return 'Jython ' + sys.platform
     elif hasattr(sys, 'pypy_version_info'):
-        v = '.'.join(map(str, sys.pypy_version_info[:3]))
+        v = '.'.join(imap(str, sys.pypy_version_info[:3]))
         if sys.pypy_version_info[3:]:
-            v += '-' + ''.join(map(str, sys.pypy_version_info[3:]))
+            v += '-' + ''.join(imap(str, sys.pypy_version_info[3:]))
         return 'PyPy ' + v
     else:
         return 'CPython'

celery/result.py

@@ -12,7 +12,7 @@ import time
 
 from collections import deque
 from copy import copy
-from future_builtins import map
+from itertools import imap
 
 from . import current_app
 from . import states
@@ -392,7 +392,7 @@ class ResultSet(ResultBase):
         :returns: the number of tasks completed.
 
         """
-        return sum(map(int, (result.successful() for result in self.results)))
+        return sum(imap(int, (result.successful() for result in self.results)))
 
     def forget(self):
         """Forget about (and possible remove the result of) all the tasks."""

celery/security/serialization.py

@@ -10,7 +10,7 @@ from __future__ import absolute_import
 
 import base64
 
-from future_builtins import zip
+from itertools import izip
 from kombu.serialization import registry, encode, decode
 from kombu.utils.encoding import bytes_to_str, str_to_bytes
 
@@ -72,7 +72,7 @@ class SecureSerializer(object):
     def _unpack(self, payload, sep='\x00\x01',
             fields=('signer', 'signature', 'content_type',
                     'content_encoding', 'body')):
-        return dict(zip(fields, b64decode(payload).split(sep)))
+        return dict(izip(fields, b64decode(payload).split(sep)))
 
 
 def register_auth(key=None, cert=None, store=None, digest='sha1',

celery/utils/text.py

@@ -8,7 +8,7 @@
 """
 from __future__ import absolute_import
 
-from future_builtins import filter, map
+from itertools import ifilter, imap
 from textwrap import fill
 
 from pprint import pformat
@@ -19,7 +19,7 @@ def dedent_initial(s, n=4):
 
 
 def dedent(s, n=4):
-    return '\n'.join(map(dedent_initial, s.splitlines()))
+    return '\n'.join(imap(dedent_initial, s.splitlines()))
 
 
 def fill_paragraphs(s, width):
@@ -27,7 +27,7 @@ def fill_paragraphs(s, width):
 
 
 def join(l):
-    return '\n'.join(filter(None, l))
+    return '\n'.join(ifilter(None, l))
 
 
 def ensure_2lines(s):

celery/utils/timer2.py

@@ -15,15 +15,14 @@ import sys
 import threading
 
 from functools import wraps
-from future_builtins import map
-from itertools import count
+from itertools import count, imap
 from time import time, sleep, mktime
 
 from datetime import datetime, timedelta
 from kombu.log import get_logger
 
 VERSION = (1, 0, 0)
-__version__ = '.'.join(map(str, VERSION))
+__version__ = '.'.join(imap(str, VERSION))
 __author__ = 'Ask Solem'
 __contact__ = 'ask@celeryproject.org'
 __homepage__ = 'http://github.com/ask/timer2/'

celery/utils/timeutils.py

@@ -8,12 +8,12 @@
 """
 from __future__ import absolute_import
 
-from kombu.utils import cached_property
+from itertools import izip
 
 from datetime import datetime, timedelta
 from dateutil import tz
 from dateutil.parser import parse as parse_iso8601
-from future_builtins import zip
+from kombu.utils import cached_property
 
 from celery.exceptions import ImproperlyConfigured
 
@@ -26,7 +26,7 @@ except ImportError:     # pragma: no cover
 
 
 DAYNAMES = 'sun', 'mon', 'tue', 'wed', 'thu', 'fri', 'sat'
-WEEKDAYS = dict(zip(DAYNAMES, range(7)))
+WEEKDAYS = dict(izip(DAYNAMES, range(7)))
 
 RATE_MODIFIER_MAP = {'s': lambda n: n,
                      'm': lambda n: n / 60.0,

celery/worker/control.py

@@ -10,9 +10,8 @@ from __future__ import absolute_import
 
 import os
 
-from future_builtins import map
-
 from kombu.utils.encoding import safe_repr
+from itertools import imap
 
 from celery.platforms import signals as _signals
 from celery.utils import timeutils
@@ -166,7 +165,7 @@ def dump_reserved(panel, safe=False, **kwargs):
         logger.debug('--Empty queue--')
         return []
     logger.debug('* Dump of currently reserved tasks:\n%s',
-                 '\n'.join(map(safe_repr, reserved)))
+                 '\n'.join(imap(safe_repr, reserved)))
     return [request.info(safe=safe)
             for request in reserved]
 
@@ -204,7 +203,7 @@ def dump_tasks(panel, taskinfoitems=None, **kwargs):
                         for field in taskinfoitems
                             if getattr(task, field, None) is not None)
         if fields:
-            info = map('='.join, fields.iteritems())
+            info = imap('='.join, fields.iteritems())
             return '{0} [{1}]'.format(task.name, ' '.join(info))
         return task.name