Python 3.7+ compatibility

celery/backends/redis.py

@@ -10,9 +10,6 @@
 
 from functools import partial
 
-from kombu.utils import cached_property, retry_over_time
-from kombu.utils.url import _parse_url
-
 from celery import states
 from celery.canvas import maybe_signature
 from celery.exceptions import ChordError, ImproperlyConfigured
@@ -21,13 +18,16 @@
 from celery.utils.functional import dictfilter
 from celery.utils.log import get_logger
 from celery.utils.timeutils import humanize_seconds
+from kombu.utils import cached_property, retry_over_time
+from kombu.utils.url import _parse_url
 
 from .base import KeyValueStoreBackend
 
 try:
     import redis
     from redis.exceptions import ConnectionError
-    from kombu.transport.redis import get_redis_error_classes
+
+    from .patched_kombu_redis_transport import get_redis_error_classes
 except ImportError:                 # pragma: no cover
     redis = None                    # noqa
     ConnectionError = None          # noqa

celery/concurrency/asynpool.py

@@ -19,38 +19,46 @@
 from __future__ import absolute_import
 
 import errno
+import gc
+import importlib
 import os
 import select
 import socket
 import struct
 import sys
 import time
-
 from collections import deque, namedtuple
 from io import BytesIO
 from pickle import HIGHEST_PROTOCOL
 from time import sleep
 from weakref import WeakValueDictionary, ref
 
 from amqp.utils import promise
-from billiard.pool import RUN, TERMINATE, ACK, NACK, WorkersJoined
 from billiard import pool as _pool
-from billiard.compat import buf_t, setblocking, isblocking
+from billiard.compat import buf_t, isblocking, setblocking
 from billiard.einfo import ExceptionInfo
+from billiard.pool import ACK, NACK, RUN, TERMINATE, WorkersJoined
 from billiard.queues import _SimpleQueue
-from kombu.async import READ, WRITE, ERR
-from kombu.serialization import pickle as _pickle
-from kombu.utils import fxrange
-from kombu.utils.compat import get_errno
-from kombu.utils.eventio import SELECT_BAD_FD
+
+kombu_async = importlib.import_module("kombu.async")
+
+READ = getattr(kombu_async, 'READ')
+WRITE = getattr(kombu_async, 'WRITE')
+ERR = getattr(kombu_async, 'ERR')
+
 from celery.five import Counter, items, string_t, text_t, values
 from celery.utils.log import get_logger
 from celery.utils.text import truncate
 from celery.worker import state as worker_state
+from kombu.serialization import pickle as _pickle
+from kombu.utils import fxrange
+from kombu.utils.compat import get_errno
+from kombu.utils.eventio import SELECT_BAD_FD
 
 try:
-    from _billiard import read as __read__
     from struct import unpack_from as _unpack_from
+
+    from _billiard import read as __read__
     memoryview = memoryview
     readcanbuf = True
 

celery/utils/timer2.py

@@ -8,15 +8,20 @@
 """
 from __future__ import absolute_import
 
+import importlib
 import os
 import sys
 import threading
-
 from itertools import count
 from time import sleep
 
 from celery.five import THREAD_TIMEOUT_MAX
-from kombu.async.timer import Entry, Timer as Schedule, to_timestamp, logger
+
+kombu_async_timer = importlib.import_module("kombu.async.timer")
+Entry = getattr(kombu_async_timer, 'Entry')
+Schedule = getattr(kombu_async_timer, 'Timer')
+logger = getattr(kombu_async_timer, 'logger')
+to_timestamp = getattr(kombu_async_timer, 'to_timestamp')
 
 TIMER_DEBUG = os.environ.get('TIMER_DEBUG')
 

celery/worker/autoscale.py

@@ -13,18 +13,20 @@
 """
 from __future__ import absolute_import
 
+import importlib
 import os
 import threading
-
 from time import sleep
 
-from kombu.async.semaphore import DummyLock
-
+kombu_async_semaphore = importlib.import_module("kombu.async.semaphore")
 from celery import bootsteps
 from celery.five import monotonic
 from celery.utils.log import get_logger
 from celery.utils.threads import bgThread
 
+DummyLock = getattr(kombu_async_semaphore, 'DummyLock')
+
+
 from . import state
 from .components import Pool
 

celery/worker/components.py

@@ -9,18 +9,26 @@
 from __future__ import absolute_import
 
 import atexit
+import importlib
 import warnings
 
-from kombu.async import Hub as _Hub, get_event_loop, set_event_loop
-from kombu.async.semaphore import DummyLock, LaxBoundedSemaphore
-from kombu.async.timer import Timer as _Timer
-
+kombu_async = importlib.import_module("kombu.async")
+kombu_async_semaphore = importlib.import_module("kombu.async.semaphore")
+kombu_async_timer = importlib.import_module("kombu.async.timer")
 from celery import bootsteps
 from celery._state import _set_task_join_will_block
 from celery.exceptions import ImproperlyConfigured
 from celery.five import string_t
 from celery.utils.log import worker_logger as logger
 
+_Hub = getattr(kombu_async, 'Hub')
+get_event_loop = getattr(kombu_async, 'get_event_loop')
+set_event_loop = getattr(kombu_async, 'set_event_loop')
+DummyLock = getattr(kombu_async_semaphore, 'DummyLock')
+LaxBoundedSemaphore = getattr(kombu_async_semaphore, 'LaxBoundedSemaphore')
+_Timer = getattr(kombu_async_timer, 'Timer')
+
+
 __all__ = ['Timer', 'Hub', 'Queues', 'Pool', 'Beat', 'StateDB', 'Consumer']
 
 ERR_B_GREEN = """\

celery/worker/consumer.py

@@ -11,27 +11,23 @@
 from __future__ import absolute_import
 
 import errno
-import kombu
+import importlib
 import logging
 import os
 import socket
-
 from collections import defaultdict
 from functools import partial
 from heapq import heappush
 from operator import itemgetter
 from time import sleep
 
+import kombu
 from billiard.common import restart_state
 from billiard.exceptions import RestartFreqExceeded
-from kombu.async.semaphore import DummyLock
-from kombu.common import QoS, ignore_errors
-from kombu.syn import _detect_environment
-from kombu.utils.compat import get_errno
-from kombu.utils.encoding import safe_repr, bytes_t
-from kombu.utils.limits import TokenBucket
-
 from celery import bootsteps
+
+kombu_async_semephore = importlib.import_module("kombu.async.semaphore")
+from celery import bootsteps, chain
 from celery.app.trace import build_tracer
 from celery.canvas import signature
 from celery.exceptions import InvalidTaskError
@@ -40,9 +36,16 @@
 from celery.utils.log import get_logger
 from celery.utils.text import truncate
 from celery.utils.timeutils import humanize_seconds, rate
+from kombu.common import QoS, ignore_errors
+from kombu.syn import _detect_environment
+from kombu.utils.compat import get_errno
+from kombu.utils.encoding import bytes_t, safe_repr
+from kombu.utils.limits import TokenBucket
+
+DummyLock = getattr(kombu_async_semephore, 'DummyLock')
 
 from . import heartbeat, loops, pidbox
-from .state import task_reserved, maybe_shutdown, revoked, reserved_requests
+from .state import maybe_shutdown, reserved_requests, revoked, task_reserved
 
 try:
     buffer_t = buffer

celery/worker/strategy.py

@@ -8,13 +8,16 @@
 """
 from __future__ import absolute_import
 
+import importlib
 import logging
 
-from kombu.async.timer import to_timestamp
-from kombu.utils.encoding import safe_repr
-
+kombu_async = importlib.import_module("kombu.async")
 from celery.utils.log import get_logger
 from celery.utils.timeutils import timezone
+from kombu.utils.encoding import safe_repr
+
+kombu_async_timer = importlib.import_module("kombu.async.timer")
+to_timestamp = getattr(kombu_async_timer, 'to_timestamp')
 
 from .job import Request
 from .state import task_reserved

requirements/default.txt

@@ -1,3 +1,3 @@
-pytz>dev
+pytz>2013.6
 billiard>=3.3.0.21,<3.4
 kombu>=3.0.29,<3.1