flake8

celery/__init__.py

@@ -68,19 +68,21 @@ new_module.__dict__.update({
     "__package__": package,
     "VERSION": VERSION})
 
+
 def Celery(*args, **kwargs):
     from .app import App
     return App(*args, **kwargs)
 
+
 def _get_current_app():
     from .app import current_app
     return current_app()
 current_app = Proxy(_get_current_app)
 
+
 def bugreport():
     return current_app.bugreport()
 
 new_module.Celery = Celery
 new_module.current_app = current_app
 new_module.bugreport = bugreport
-

celery/app/__init__.py

@@ -53,10 +53,10 @@ def _app_or_default_trace(app=None):  # pragma: no cover
     from traceback import print_stack
     from multiprocessing import current_process
     if app is None:
-        if getattr(_tls, "current_app", None):
+        if getattr(state._tls, "current_app", None):
             print("-- RETURNING TO CURRENT APP --")  # noqa+
             print_stack()
-            return _tls.current_app
+            return state._tls.current_app
         if current_process()._name == "MainProcess":
             raise Exception("DEFAULT APP")
         print("-- RETURNING TO DEFAULT APP --")      # noqa+
@@ -64,6 +64,7 @@ def _app_or_default_trace(app=None):  # pragma: no cover
         return state.default_app
     return app
 
+
 def enable_trace():
     global app_or_default
     app_or_default = _app_or_default_trace

celery/app/state.py

@@ -2,6 +2,8 @@ from __future__ import absolute_import
 
 import threading
 
+default_app = None
+
 
 class _TLS(threading.local):
     #: Apps with the :attr:`~celery.app.base.BaseApp.set_as_current` attribute
@@ -14,9 +16,6 @@ class _TLS(threading.local):
 _tls = _TLS()
 
 
-default_app = None
-
-
 def set_default_app(app):
     global default_app
     default_app = app
@@ -28,42 +27,3 @@ def current_app():
 
 def current_task():
     return getattr(_tls, "current_task", None)
-
-
-def _app_or_default(app=None):
-    """Returns the app provided or the default app if none.
-
-    The environment variable :envvar:`CELERY_TRACE_APP` is used to
-    trace app leaks.  When enabled an exception is raised if there
-    is no active app.
-
-    """
-    if app is None:
-        return getattr(_tls, "current_app", None) or default_app
-    return app
-
-
-def _app_or_default_trace(app=None):  # pragma: no cover
-    from traceback import print_stack
-    from multiprocessing import current_process
-    if app is None:
-        if getattr(_tls, "current_app", None):
-            print("-- RETURNING TO CURRENT APP --")  # noqa+
-            print_stack()
-            return _tls.current_app
-        if current_process()._name == "MainProcess":
-            raise Exception("DEFAULT APP")
-        print("-- RETURNING TO DEFAULT APP --")      # noqa+
-        print_stack()
-        return default_app
-    return app
-
-app_or_default = None
-def enable_trace():
-    global app_or_default
-    app_or_default = _app_or_default_trace
-
-
-def disable_trace():
-    global app_or_default
-    app_or_default = _app_or_default

celery/backends/__init__.py

@@ -25,6 +25,9 @@ BACKEND_ALIASES = {
     "disabled": "celery.backends.base:DisabledBackend",
 }
 
+#: deprecated alias to ``current_app.backend``.
+default_backend = Proxy(lambda: current_app.backend)
+
 
 @memoize(100)
 def get_backend_cls(backend=None, loader=None):
@@ -45,8 +48,3 @@ def get_backend_by_url(backend=None, loader=None):
         url = backend
         backend, _, _, _, _, _, _ = _parse_url(url)
     return get_backend_cls(backend, loader), url
-
-
-
-# deprecate this.
-default_backend = Proxy(lambda: current_app.backend)

celery/bin/celeryctl.py

@@ -5,11 +5,7 @@ from __future__ import with_statement
 if __name__ == "__main__" and __package__ is None:
     __package__ = "celery.bin.celeryctl"
 
-import sys
+from .celery import CeleryCommand as celeryctl, main  # noqa
 
-from ..platforms import EX_FAILURE
-
-from .celery import CeleryCommand as celeryctl, main
-
-if __name__ == "__main__":          # pragma: no cover
+if __name__ == "__main__":  # pragma: no cover
     main()

celery/concurrency/base.py

@@ -5,11 +5,13 @@ import logging
 import os
 import time
 
+from kombu.log import anon_logger
 from kombu.utils.encoding import safe_repr
 
-from .. import log
 from ..utils import timer2
 
+_default_logger = anon_logger("celery.concurrency")
+
 
 def apply_target(target, args=(), kwargs={}, callback=None,
         accept_callback=None, pid=None, **_):
@@ -47,7 +49,7 @@ class BasePool(object):
     def __init__(self, limit=None, putlocks=True, logger=None, **options):
         self.limit = limit
         self.putlocks = putlocks
-        self.logger = logger or log.get_default_logger()
+        self.logger = logger or _default_logger
         self.options = options
         self._does_debug = self.logger.isEnabledFor(logging.DEBUG)
 

celery/task/sets.py

@@ -17,7 +17,7 @@ from itertools import chain
 from kombu.utils import reprcall
 
 from .. import current_app
-from ..app import app_or_default, current_task
+from ..app import current_task
 from ..datastructures import AttributeDict
 from ..utils import cached_property, uuid
 from ..utils.functional import maybe_list

celery/tests/test_app/__init__.py

@@ -2,7 +2,6 @@ from __future__ import absolute_import
 from __future__ import with_statement
 
 import os
-import sys
 
 from mock import Mock
 

celery/tests/test_app/test_log.py

@@ -6,7 +6,6 @@ import logging
 from tempfile import mktemp
 
 from celery import current_app
-from celery import log
 from celery.utils.log import LoggingProxy
 from celery.utils import uuid
 from celery.utils.compat import _CompatLoggerAdapter

celery/tests/test_worker/test_worker_job.py

@@ -3,7 +3,6 @@ from __future__ import absolute_import
 from __future__ import with_statement
 
 import anyjson
-import logging
 import os
 import sys
 import time

celery/utils/compat.py

@@ -278,24 +278,3 @@ else:
                         stat = os.stat(self.baseFilename)
                     self.dev, self.ino = stat[ST_DEV], stat[ST_INO]
                 logging.FileHandler.emit(self, record)
-
-
-
-def add_compat_modules():
-
-    from celery import current_app
-    from celery.local import Proxy
-
-    from types import ModuleType
-
-    class messaging(ModuleType):
-        TaskPublisher = Proxy(lambda: current_app.amqp.TaskPublisher)
-        ConsumerSet = Proxy(lambda: current_app.amqp.ConsumerSet)
-        TaskConsumer = Proxy(lambda: current_app.amqp.TaskConsumer)
-        establish_connection = Proxy(lambda: current_app.broker_connection)
-        with_connection = Proxy(lambda: current_app.with_default_connection)
-        get_consumer_set = Proxy(lambda: current_app.amqp.get_task_consumer)
-
-    sys.modules["celery.messaging"] = messaging("celery.messaging")
-
-