Makes it possible to set .app on some classes

celery/app/task/__init__.py

@@ -100,7 +100,7 @@ class TaskType(type):
         # in the '_app' attribute.  Previously this was a regular attribute,
         # so we should support classes defining it.
         _app1, _app2 = attrs.pop("_app", None), attrs.pop("app", None)
-        app = attrs["_app"] =  _app1 or _app2 or current_app
+        app = attrs["_app"] = _app1 or _app2 or current_app
 
         # - Automatically generate missing/empty name.
         autoname = False

celery/apps/beat.py

@@ -27,6 +27,7 @@ Configuration ->
 class Beat(configurated):
     Service = beat.Service
 
+    app = None
     loglevel = from_config("log_level")
     logfile = from_config("log_file")
     schedule = from_config("schedule_filename")
@@ -37,7 +38,7 @@ class Beat(configurated):
     def __init__(self, max_interval=None, app=None,
             socket_timeout=30, pidfile=None, **kwargs):
         """Starts the celerybeat task scheduler."""
-        self.app = app = app_or_default(app)
+        self.app = app = app_or_default(app or self.app)
         self.setup_defaults(kwargs, namespace="celerybeat")
 
         self.max_interval = max_interval

celery/apps/worker.py

@@ -75,6 +75,7 @@ def get_process_name():
 class Worker(configurated):
     WorkController = WorkController
 
+    app = None
     inherit_confopts = (WorkController, )
     loglevel = from_config("log_level")
     redirect_stdouts = from_config()
@@ -83,7 +84,7 @@ class Worker(configurated):
     def __init__(self, hostname=None, discard=False, embed_clockservice=False,
             queues=None, include=None, app=None, pidfile=None,
             autoscale=None, autoreload=False, **kwargs):
-        self.app = app = app_or_default(app)
+        self.app = app = app_or_default(app or self.app)
         self.hostname = hostname or socket.gethostname()
 
         # this signal can be used to set up configuration for

celery/result.py

@@ -46,6 +46,7 @@ class AsyncResult(object):
     :keyword backend: see :attr:`backend`.
 
     """
+    app = None
 
     #: Error raised for timeouts.
     TimeoutError = TimeoutError
@@ -61,7 +62,7 @@ class AsyncResult(object):
 
     def __init__(self, id, backend=None, task_name=None,
             app=None, parent=None):
-        self.app = app_or_default(app)
+        self.app = app_or_default(app or self.app)
         self.id = id
         self.backend = backend or self.app.backend
         self.task_name = task_name
@@ -293,12 +294,13 @@ class ResultSet(object):
     :param results: List of result instances.
 
     """
+    app = None
 
     #: List of results in in the set.
     results = None
 
     def __init__(self, results, app=None, **kwargs):
-        self.app = app_or_default(app)
+        self.app = app_or_default(app or self.app)
         self.results = results
 
     def add(self, result):

celery/task/__init__.py

@@ -12,7 +12,7 @@
 from __future__ import absolute_import
 
 from .. import current_app
-from ..app import app_or_default, current_task as _current_task
+from ..app import current_task as _current_task
 from ..local import Proxy
 from ..utils import uuid
 

celery/task/sets.py

@@ -146,9 +146,10 @@ class group(UserList):
         >>> list_of_return_values = group_result.join()  # *expensive*
 
     """
+    _app = None
 
     def __init__(self, tasks=None, app=None, Publisher=None):
-        self._app = app
+        self._app = app or self._app
         self.data = [maybe_subtask(t) for t in tasks or []]
         self._Publisher = Publisher
 

celery/worker/__init__.py

@@ -169,6 +169,7 @@ class WorkController(configurated):
     CLOSE = CLOSE
     TERMINATE = TERMINATE
 
+    app = None
     concurrency = from_config()
     loglevel = logging.ERROR
     logfile = from_config("log_file")
@@ -197,7 +198,7 @@ class WorkController(configurated):
     def __init__(self, loglevel=None, hostname=None, logger=None,
             ready_callback=noop,
             queues=None, app=None, **kwargs):
-        self.app = app_or_default(app)
+        self.app = app_or_default(app or self.app)
         self._shutdown_complete = threading.Event()
         self.setup_defaults(kwargs, namespace="celeryd")
         self.app.select_queues(queues)  # select queues subset.