Celery now supports Django out of the box, just by setting DJANGO_SETTINGS_MODULE envvar

There's a new example Django project in examples/django

Closes #795

celery/app/base.py

@@ -35,6 +35,10 @@ from .defaults import DEFAULTS, find_deprecated_settings
 from .registry import TaskRegistry
 from .utils import AppPickler, Settings, bugreport, _unpickle_app
 
+DEFAULT_FIXUPS = (
+    'celery.fixups.django:DjangoFixup',
+)
+
 
 def _unpickle_appattr(reverse_name, args):
     """Given an attribute name and a list of args, gets
@@ -92,6 +96,8 @@ class Celery(object):
             self._preconf['BROKER_URL'] = broker
         if include:
             self._preconf['CELERY_IMPORTS'] = include
+        self.fixups = list(filter(None, (symbol_by_name(f).include(self)
+                                        for f in DEFAULT_FIXUPS)))
 
         if self.set_as_current:
             self.set_current()
@@ -195,6 +201,9 @@ class Celery(object):
     def config_from_cmdline(self, argv, namespace='celery'):
         self.conf.update(self.loader.cmdline_config_parser(argv, namespace))
 
+    def autodiscover_tasks(self, packages, related_name='tasks'):
+        self.loader.autodiscover_tasks(packages, related_name)
+
     def send_task(self, name, args=None, kwargs=None, countdown=None,
             eta=None, task_id=None, producer=None, connection=None,
             result_cls=None, expires=None, queues=None, publisher=None,

celery/apps/worker.py

@@ -102,7 +102,6 @@ class Worker(WorkController):
         )
 
     def on_init_namespace(self):
-        print('SETUP LOGGING: %r' % (self.redirect_stdouts, ))
         self.setup_logging()
 
     def on_start(self):

celery/bin/base.py

@@ -306,15 +306,20 @@ class Command(object):
 
     def find_app(self, app):
         try:
+            print('sym by name: %r' % (app, ))
             sym = self.symbol_by_name(app)
         except AttributeError:
+            print('ATTRIBUTE ERROR')
             # last part was not an attribute, but a module
             sym = import_from_cwd(app)
         if isinstance(sym, ModuleType):
-            if getattr(sym, '__path__', None):
-                return self.find_app('{0}.celery:'.format(
-                            app.replace(':', '')))
-            return sym.celery
+            try:
+                return sym.celery
+            except AttributeError:
+                if getattr(sym, '__path__', None):
+                    return self.find_app('{0}.celery:'.format(
+                                app.replace(':', '')))
+                raise
         return sym
 
     def symbol_by_name(self, name):

celery/fixups/django.py

@@ -0,0 +1,185 @@
+from __future__ import absolute_import
+
+import os
+import sys
+import warnings
+
+from datetime import datetime
+
+from celery import signals
+from celery.utils.imports import import_from_cwd
+
+SETTINGS_MODULE = os.environ.get('DJANGO_SETTINGS_MODULE')
+
+
+def _maybe_close_fd(fh):
+    try:
+        os.close(fh.fileno())
+    except (AttributeError, OSError, TypeError):
+        # TypeError added for celery#962
+        pass
+
+
+class DjangoFixup(object):
+    _db_recycles = 0
+
+    @classmethod
+    def include(cls, app):
+        if SETTINGS_MODULE:
+            self = cls(app)
+            self.install()
+            return self
+
+    def __init__(self, app):
+        from django import db
+        from django.core import cache
+        from django.conf import settings
+        from django.core.mail import mail_admins
+
+        # Current time and date
+        try:
+            from django.utils.timezone import now
+        except ImportError:  # pre django-1.4
+            now = datetime.now  # noqa
+
+        # Database-related exceptions.
+        from django.db import DatabaseError
+        try:
+            import MySQLdb as mysql
+            _my_database_errors = (mysql.DatabaseError,
+                                   mysql.InterfaceError,
+                                   mysql.OperationalError)
+        except ImportError:
+            _my_database_errors = ()      # noqa
+        try:
+            import psycopg2 as pg
+            _pg_database_errors = (pg.DatabaseError,
+                                   pg.InterfaceError,
+                                   pg.OperationalError)
+        except ImportError:
+            _pg_database_errors = ()      # noqa
+        try:
+            import sqlite3
+            _lite_database_errors = (sqlite3.DatabaseError,
+                                     sqlite3.InterfaceError,
+                                     sqlite3.OperationalError)
+        except ImportError:
+            _lite_database_errors = ()    # noqa
+        try:
+            import cx_Oracle as oracle
+            _oracle_database_errors = (oracle.DatabaseError,
+                                       oracle.InterfaceError,
+                                       oracle.OperationalError)
+        except ImportError:
+            _oracle_database_errors = ()  # noqa
+
+        self.app = app
+        self.db_reuse_max = self.app.conf.get('CELERY_DB_REUSE_MAX', None)
+        self._cache = cache
+        self._settings = settings
+        self._db = db
+        self._mail_admins = mail_admins
+        self._now = now
+        self.database_errors = (
+            (DatabaseError, ) +
+            _my_database_errors +
+            _pg_database_errors +
+            _lite_database_errors +
+            _oracle_database_errors,
+        )
+
+    def install(self):
+        sys.path.append(os.getcwd())
+        signals.beat_embedded_init.connect(self.close_database)
+        signals.worker_ready.connect(self.on_worker_ready)
+        signals.task_prerun.connect(self.on_task_prerun)
+        signals.task_postrun.connect(self.on_task_postrun)
+        signals.worker_init.connect(self.on_worker_init)
+        signals.worker_process_init.connect(self.on_worker_process_init)
+
+        self.app.loader.now = self.now
+        self.app.loader.mail_admins = self.mail_admins
+
+    def now(self, utc=False):
+        return datetime.utcnow() if utc else self._now()
+
+    def mail_admins(self, subject, body, fail_silently=False, **kwargs):
+        return self._mail_admins(subject, body, fail_silently=fail_silently)
+
+    def on_worker_init(self, **kwargs):
+        """Called when the worker starts.
+
+        Automatically discovers any ``tasks.py`` files in the applications
+        listed in ``INSTALLED_APPS``.
+
+        """
+        self.close_database()
+        self.close_cache()
+
+    def on_worker_process_init(self, **kwargs):
+        # the parent process may have established these,
+        # so need to close them.
+
+        # calling db.close() on some DB connections will cause
+        # the inherited DB conn to also get broken in the parent
+        # process so we need to remove it without triggering any
+        # network IO that close() might cause.
+        try:
+            for c in self._db.connections.all():
+                if c and c.connection:
+                    _maybe_close_fd(c.connection)
+        except AttributeError:
+            if self._db.connection and self._db.connection.connection:
+                _maybe_close_fd(self._db.connection.connection)
+
+        # use the _ version to avoid DB_REUSE preventing the conn.close() call
+        self._close_database()
+        self.close_cache()
+
+    def on_task_prerun(self, sender, **kwargs):
+        """Called before every task."""
+        if not getattr(sender.request, 'is_eager', False):
+            self.close_database()
+
+    def on_task_postrun(self, **kwargs):
+        """Does everything necessary for Django to work in a long-living,
+        multiprocessing environment.
+
+        """
+        # See http://groups.google.com/group/django-users/
+        #            browse_thread/thread/78200863d0c07c6d/
+        self.close_database()
+        self.close_cache()
+
+    def close_database(self, **kwargs):
+        if not self.db_reuse_max:
+            return self._close_database()
+        if self._db_recycles >= self.db_reuse_max * 2:
+            self._db_recycles = 0
+            self._close_database()
+        self._db_recycles += 1
+
+    def _close_database(self):
+        try:
+            funs = [conn.close for conn in self._db.connections]
+        except AttributeError:
+            funs = [self._db.close_connection]  # pre multidb
+
+        for close in funs:
+            try:
+                close()
+            except self.database_errors, exc:
+                str_exc = str(exc)
+                if 'closed' not in str_exc and 'not connected' not in str_exc:
+                    raise
+
+    def close_cache(self):
+        try:
+            self._cache.cache.close()
+        except (TypeError, AttributeError):
+            pass
+
+    def on_worker_ready(self, **kwargs):
+        if self._settings.DEBUG:
+            warnings.warn('Using settings.DEBUG leads to a memory leak, never '
+                          'use this setting in production environments!')

celery/loaders/base.py

@@ -9,6 +9,7 @@
 from __future__ import absolute_import
 
 import anyjson
+import imp
 import importlib
 import os
 import re
@@ -32,6 +33,8 @@ and as such the configuration could not be loaded.
 Please set this variable and make it point to
 a configuration module.""")
 
+_RACE_PROTECTION = False
+
 
 class BaseLoader(object):
     """The base class for loaders.
@@ -209,6 +212,11 @@ class BaseLoader(object):
     def read_configuration(self):
         return {}
 
+    def autodiscover_tasks(self, packages, related_name='tasks'):
+        self.task_modules.update(mod.__name__
+            for mod in autodiscover_tasks(packages, related_name) if mod
+        )
+
     @property
     def conf(self):
         """Loader configuration."""
@@ -219,3 +227,32 @@ class BaseLoader(object):
     @cached_property
     def mail(self):
         return self.import_module('celery.utils.mail')
+
+
+def autodiscover_tasks(packages, related_name='tasks'):
+    global _RACE_PROTECTION
+
+    if _RACE_PROTECTION:
+        return
+    _RACE_PROTECTION = True
+    try:
+        return [find_related_module(pkg, related_name) for pkg in packages]
+    finally:
+        _RACE_PROTECTION = False
+
+
+def find_related_module(package, related_name):
+    """Given a package name and a module name, tries to find that
+    module."""
+
+    try:
+        pkg_path = importlib.import_module(package).__path__
+    except AttributeError:
+        return
+
+    try:
+        imp.find_module(related_name, pkg_path)
+    except ImportError:
+        return
+
+    return importlib.import_module('{0}.{1}'.format(package, related_name))

celery/task/trace.py

@@ -36,11 +36,8 @@ from celery.utils.log import get_logger
 _logger = get_logger(__name__)
 
 send_prerun = signals.task_prerun.send
-prerun_receivers = signals.task_prerun.receivers
 send_postrun = signals.task_postrun.send
-postrun_receivers = signals.task_postrun.receivers
 send_success = signals.task_success.send
-success_receivers = signals.task_success.receivers
 STARTED = states.STARTED
 SUCCESS = states.SUCCESS
 IGNORED = states.IGNORED
@@ -198,6 +195,10 @@ def build_tracer(name, task, loader=None, hostname=None, store_errors=True,
     pop_task = _task_stack.pop
     on_chord_part_return = backend.on_chord_part_return
 
+    prerun_receivers = signals.task_prerun.receivers
+    postrun_receivers = signals.task_postrun.receivers
+    success_receivers = signals.task_success.receivers
+
     from celery import canvas
     subtask = canvas.subtask
 

docs/reference/celery.rst

@@ -134,6 +134,27 @@ Application
             >>> os.environ["CELERY_CONFIG_MODULE"] = "myapp.celeryconfig"
             >>> celery.config_from_envvar("CELERY_CONFIG_MODULE")
 
+    .. method:: Celery.autodiscover_tasks(packages, related_name="tasks")
+
+        With a list of packages, try to import modules of a specific name (by
+        default 'tasks').
+
+        For example if you have an (imagined) directory tree like this::
+
+            foo/__init__.py
+               tasks.py
+               models.py
+
+            bar/__init__.py
+                tasks.py
+                models.py
+
+            baz/__init__.py
+                models.py
+
+        Then calling ``app.autodiscover_tasks(['foo', bar', 'baz'])`` will
+        result in the modules ``foo.tasks`` and ``bar.tasks`` being imported.
+
     .. method:: Celery.add_defaults(d)
 
         Add default configuration from dict ``d``.

examples/django/proj/README.rst

@@ -0,0 +1,36 @@
+==============================================================
+ Example Django project using Celery
+==============================================================
+
+Contents
+========
+
+:file:`proj/`
+-------------
+
+This is the project iself, created using
+:program:`django-admin.py startproject proj`, and then the settings module
+(:file:`proj/settings.py`) was modified to add ``tasks`` and ``demoapp`` to
+``INSTALLED_APPS``
+
+:file:`tasks/`
+--------------
+
+This app contains the Celery application instance for this project,
+we take configuration from Django settings and use ``autodiscover_tasks`` to
+find task modules inside all packages listed in ``INSTALLED_APPS``.
+
+:file:`demoapp/`
+----------------
+
+Example generic app.  This is decoupled from the rest of the project by using
+the ``@shared_task`` decorator.  Shared tasks are shared between all Celery
+instances.
+
+
+Starting the worker
+===================
+
+.. code-block:: bash
+
+    $ DJANGO_SETTINGS_MODULE='proj.settings' celery -A tasks worker -l info

examples/django/proj/demoapp/models.py

@@ -0,0 +1,3 @@
+from django.db import models
+
+# Create your models here.

examples/django/proj/demoapp/tasks.py

@@ -0,0 +1,16 @@
+from celery import shared_task
+
+
+@shared_task
+def add(x, y):
+    return x + y
+
+
+@shared_task
+def mul(x, y):
+    return x * y
+
+
+@shared_task
+def xsum(numbers):
+    return sum(numbers)

examples/django/proj/demoapp/tests.py

@@ -0,0 +1,16 @@
+"""
+This file demonstrates writing tests using the unittest module. These will pass
+when you run "manage.py test".
+
+Replace this with more appropriate tests for your application.
+"""
+
+from django.test import TestCase
+
+
+class SimpleTest(TestCase):
+    def test_basic_addition(self):
+        """
+        Tests that 1 + 1 always equals 2.
+        """
+        self.assertEqual(1 + 1, 2)

examples/django/proj/demoapp/views.py

@@ -0,0 +1 @@
+# Create your views here.

examples/django/proj/manage.py

@@ -0,0 +1,10 @@
+#!/usr/bin/env python
+import os
+import sys
+
+if __name__ == "__main__":
+    os.environ.setdefault("DJANGO_SETTINGS_MODULE", "proj.settings")
+
+    from django.core.management import execute_from_command_line
+
+    execute_from_command_line(sys.argv)

examples/django/proj/proj/settings.py

@@ -0,0 +1,153 @@
+# Django settings for proj project.
+
+DEBUG = True
+TEMPLATE_DEBUG = DEBUG
+
+ADMINS = (
+    # ('Your Name', 'your_email@example.com'),
+)
+
+MANAGERS = ADMINS
+
+DATABASES = {
+    'default': {
+        'ENGINE': 'django.db.backends.', # Add 'postgresql_psycopg2', 'mysql', 'sqlite3' or 'oracle'.
+        'NAME': '',                      # Or path to database file if using sqlite3.
+        'USER': '',                      # Not used with sqlite3.
+        'PASSWORD': '',                  # Not used with sqlite3.
+        'HOST': '',                      # Set to empty string for localhost. Not used with sqlite3.
+        'PORT': '',                      # Set to empty string for default. Not used with sqlite3.
+    }
+}
+
+# Local time zone for this installation. Choices can be found here:
+# http://en.wikipedia.org/wiki/List_of_tz_zones_by_name
+# although not all choices may be available on all operating systems.
+# In a Windows environment this must be set to your system time zone.
+TIME_ZONE = 'America/Chicago'
+
+# Language code for this installation. All choices can be found here:
+# http://www.i18nguy.com/unicode/language-identifiers.html
+LANGUAGE_CODE = 'en-us'
+
+SITE_ID = 1
+
+# If you set this to False, Django will make some optimizations so as not
+# to load the internationalization machinery.
+USE_I18N = True
+
+# If you set this to False, Django will not format dates, numbers and
+# calendars according to the current locale.
+USE_L10N = True
+
+# If you set this to False, Django will not use timezone-aware datetimes.
+USE_TZ = True
+
+# Absolute filesystem path to the directory that will hold user-uploaded files.
+# Example: "/home/media/media.lawrence.com/media/"
+MEDIA_ROOT = ''
+
+# URL that handles the media served from MEDIA_ROOT. Make sure to use a
+# trailing slash.
+# Examples: "http://media.lawrence.com/media/", "http://example.com/media/"
+MEDIA_URL = ''
+
+# Absolute path to the directory static files should be collected to.
+# Don't put anything in this directory yourself; store your static files
+# in apps' "static/" subdirectories and in STATICFILES_DIRS.
+# Example: "/home/media/media.lawrence.com/static/"
+STATIC_ROOT = ''
+
+# URL prefix for static files.
+# Example: "http://media.lawrence.com/static/"
+STATIC_URL = '/static/'
+
+# Additional locations of static files
+STATICFILES_DIRS = (
+    # Put strings here, like "/home/html/static" or "C:/www/django/static".
+    # Always use forward slashes, even on Windows.
+    # Don't forget to use absolute paths, not relative paths.
+)
+
+# List of finder classes that know how to find static files in
+# various locations.
+STATICFILES_FINDERS = (
+    'django.contrib.staticfiles.finders.FileSystemFinder',
+    'django.contrib.staticfiles.finders.AppDirectoriesFinder',
+#    'django.contrib.staticfiles.finders.DefaultStorageFinder',
+)
+
+# Make this unique, and don't share it with anybody.
+SECRET_KEY = 'x2$s&0z2xehpnt_99i8q3)4)t*5q@+n(+6jrqz4@rt%a8fdf+!'
+
+# List of callables that know how to import templates from various sources.
+TEMPLATE_LOADERS = (
+    'django.template.loaders.filesystem.Loader',
+    'django.template.loaders.app_directories.Loader',
+#     'django.template.loaders.eggs.Loader',
+)
+
+MIDDLEWARE_CLASSES = (
+    'django.middleware.common.CommonMiddleware',
+    'django.contrib.sessions.middleware.SessionMiddleware',
+    'django.middleware.csrf.CsrfViewMiddleware',
+    'django.contrib.auth.middleware.AuthenticationMiddleware',
+    'django.contrib.messages.middleware.MessageMiddleware',
+    # Uncomment the next line for simple clickjacking protection:
+    # 'django.middleware.clickjacking.XFrameOptionsMiddleware',
+)
+
+ROOT_URLCONF = 'proj.urls'
+
+# Python dotted path to the WSGI application used by Django's runserver.
+WSGI_APPLICATION = 'proj.wsgi.application'
+
+TEMPLATE_DIRS = (
+    # Put strings here, like "/home/html/django_templates" or "C:/www/django/templates".
+    # Always use forward slashes, even on Windows.
+    # Don't forget to use absolute paths, not relative paths.
+)
+
+INSTALLED_APPS = (
+    'django.contrib.auth',
+    'django.contrib.contenttypes',
+    'django.contrib.sessions',
+    'django.contrib.sites',
+    'django.contrib.messages',
+    'django.contrib.staticfiles',
+    'tasks',
+    'demoapp',
+    # Uncomment the next line to enable the admin:
+    # 'django.contrib.admin',
+    # Uncomment the next line to enable admin documentation:
+    # 'django.contrib.admindocs',
+)
+
+# A sample logging configuration. The only tangible logging
+# performed by this configuration is to send an email to
+# the site admins on every HTTP 500 error when DEBUG=False.
+# See http://docs.djangoproject.com/en/dev/topics/logging for
+# more details on how to customize your logging configuration.
+LOGGING = {
+    'version': 1,
+    'disable_existing_loggers': False,
+    'filters': {
+        'require_debug_false': {
+            '()': 'django.utils.log.RequireDebugFalse'
+        }
+    },
+    'handlers': {
+        'mail_admins': {
+            'level': 'ERROR',
+            'filters': ['require_debug_false'],
+            'class': 'django.utils.log.AdminEmailHandler'
+        }
+    },
+    'loggers': {
+        'django.request': {
+            'handlers': ['mail_admins'],
+            'level': 'ERROR',
+            'propagate': True,
+        },
+    }
+}

examples/django/proj/proj/urls.py

@@ -0,0 +1,17 @@
+from django.conf.urls import patterns, include, url
+
+# Uncomment the next two lines to enable the admin:
+# from django.contrib import admin
+# admin.autodiscover()
+
+urlpatterns = patterns('',
+    # Examples:
+    # url(r'^$', 'proj.views.home', name='home'),
+    # url(r'^proj/', include('proj.foo.urls')),
+
+    # Uncomment the admin/doc line below to enable admin documentation:
+    # url(r'^admin/doc/', include('django.contrib.admindocs.urls')),
+
+    # Uncomment the next line to enable the admin:
+    # url(r'^admin/', include(admin.site.urls)),
+)

examples/django/proj/proj/wsgi.py

@@ -0,0 +1,28 @@
+"""
+WSGI config for proj project.
+
+This module contains the WSGI application used by Django's development server
+and any production WSGI deployments. It should expose a module-level variable
+named ``application``. Django's ``runserver`` and ``runfcgi`` commands discover
+this application via the ``WSGI_APPLICATION`` setting.
+
+Usually you will have the standard Django WSGI application here, but it also
+might make sense to replace the whole Django WSGI application with a custom one
+that later delegates to the Django one. For example, you could introduce WSGI
+middleware here, or combine a Django application with an application of another
+framework.
+
+"""
+import os
+
+os.environ.setdefault("DJANGO_SETTINGS_MODULE", "proj.settings")
+
+# This application object is used by any WSGI server configured to use this
+# file. This includes Django's development server, if the WSGI_APPLICATION
+# setting points here.
+from django.core.wsgi import get_wsgi_application
+application = get_wsgi_application()
+
+# Apply WSGI middleware here.
+# from helloworld.wsgi import HelloWorldApplication
+# application = HelloWorldApplication(application)

examples/django/proj/tasks/__init__.py

@@ -0,0 +1,14 @@
+from __future__ import absolute_import
+
+from celery import Celery
+from django.conf import settings
+
+
+celery = Celery('tasks', broker='amqp://localhost')
+celery.config_from_object(settings)
+celery.autodiscover_tasks(settings.INSTALLED_APPS)
+
+
+@celery.task
+def debug_task():
+    print(repr(debug_task.request))

examples/django/proj/tasks/models.py

@@ -0,0 +1,3 @@
+from django.db import models
+
+# Create your models here.

examples/django/proj/tasks/tests.py

@@ -0,0 +1,16 @@
+"""
+This file demonstrates writing tests using the unittest module. These will pass
+when you run "manage.py test".
+
+Replace this with more appropriate tests for your application.
+"""
+
+from django.test import TestCase
+
+
+class SimpleTest(TestCase):
+    def test_basic_addition(self):
+        """
+        Tests that 1 + 1 always equals 2.
+        """
+        self.assertEqual(1 + 1, 2)

examples/django/proj/tasks/views.py

@@ -0,0 +1 @@
+# Create your views here.