Merge branch 'release/0.6'

Author: jezdez

.travis.yml

@@ -12,8 +12,8 @@ install:
 before_script:
   - flake8 celery_haystack --ignore=E501
 script:
-  - coverage run --branch --source=celery_haystack --omit=celery_haystack/tests `which django-admin.py` test celery_haystack
-  - coverage report
+  - coverage run --branch --source=celery_haystack `which django-admin.py` test celery_haystack
+  - coverage report --omit=celery_haystack/test*
 env:
   - DJANGO=1.3.1 HAYSTACK=v1
   - DJANGO=1.3.1 HAYSTACK=v2

AUTHORS

@@ -1,2 +1,4 @@
+Josh Bohde
+Germán M. Bravo
 Jannis Leidel <[email protected]>
 Daniel Lindsley

LICENSE

@@ -1,4 +1,4 @@
-Copyright (c) 2011-2012, Jannis Leidel.
+Copyright (c) 2011-2012, Jannis Leidel and other contributors.
 All rights reserved.
 
 Redistribution and use in source and binary forms, with or without modification,

README.rst

@@ -2,6 +2,10 @@
 celery-haystack
 ===============
 
+.. image:: https://secure.travis-ci.org/jezdez/celery-haystack.png?branch=develop
+    :alt: Build Status
+    :target: http://travis-ci.org/jezdez/celery-haystack
+
 This Django app allows you to utilize Celery for automatically updating and
 deleting objects in a Haystack_ search index.
 
@@ -30,11 +34,13 @@ By default a few dependencies will automatically be installed:
 
 - django-appconf_ -- An app to gracefully handle application settings.
 
-- versiontools_ -- A library to help staying compatible to `PEP 386`_.
+- `django-celery-transactions`_ -- An app that "holds on to Celery tasks
+  until the current database transaction is committed, avoiding potential
+  race conditions as described in `Celery's user guide`_."
 
 .. _django-appconf: http://pypi.python.org/pypi/django-appconf
-.. _versiontools: http://pypi.python.org/pypi/versiontools
-.. _`PEP 386`: http://www.python.org/dev/peps/pep-0386/
+.. _`django-celery-transactions`: https://github.com/chrisdoble/django-celery-transactions 
+.. _`Celery's user guide`: http://celery.readthedocs.org/en/latest/userguide/tasks.html#database-transactions
 
 Setup
 -----

celery_haystack/__init__.py

@@ -1 +1 @@
-__version__ = '0.5'
+__version__ = '0.6'

celery_haystack/conf.py

@@ -8,6 +8,7 @@ class CeleryHaystack(AppConf):
     RETRY_DELAY = 5 * 60
     MAX_RETRIES = 1
     DEFAULT_TASK = 'celery_haystack.tasks.CeleryHaystackSignalHandler'
+    TRANSACTION_SAFE = True
 
     COMMAND_BATCH_SIZE = None
     COMMAND_AGE = None

celery_haystack/indexes.py

@@ -25,19 +25,27 @@ def handle_model(self, model):
     # operation.
     def _setup_save(self, model=None):
         model = self.handle_model(model)
-        signals.post_save.connect(self.enqueue_save, sender=model)
+        signals.post_save.connect(self._enqueue_save, sender=model, dispatch_uid=CelerySearchIndex)
 
     def _setup_delete(self, model=None):
         model = self.handle_model(model)
-        signals.post_delete.connect(self.enqueue_delete, sender=model)
+        signals.post_delete.connect(self._enqueue_delete, sender=model, dispatch_uid=CelerySearchIndex)
 
     def _teardown_save(self, model=None):
         model = self.handle_model(model)
-        signals.post_save.disconnect(self.enqueue_save, sender=model)
+        signals.post_save.disconnect(self._enqueue_save, sender=model, dispatch_uid=CelerySearchIndex)
 
     def _teardown_delete(self, model=None):
         model = self.handle_model(model)
-        signals.post_delete.disconnect(self.enqueue_delete, sender=model)
+        signals.post_delete.disconnect(self._enqueue_delete, sender=model, dispatch_uid=CelerySearchIndex)
+
+    def _enqueue_save(self, instance, **kwargs):
+        if not getattr(instance, 'skip_indexing', False):
+            self.enqueue_save(instance, **kwargs)
+
+    def _enqueue_delete(self, instance, **kwargs):
+        if not getattr(instance, 'skip_indexing', False):
+            self.enqueue_delete(instance, **kwargs)
 
     def enqueue_save(self, instance, **kwargs):
         return self.enqueue('update', instance)

celery_haystack/tasks.py

@@ -1,29 +1,39 @@
+import logging
 from django.core.exceptions import ImproperlyConfigured
 from django.core.management import call_command
 from django.db.models.loading import get_model
 
-from celery.task import Task
 from celery_haystack.conf import settings
 
 try:
-    from haystack import connections
-    index_holder = connections['default'].get_unified_index()
+    from haystack import connections, connection_router
     from haystack.exceptions import NotHandled as IndexNotFoundException
     legacy = False
 except ImportError:
     try:
-        from haystack import site as index_holder
+        from haystack import site
         from haystack.exceptions import NotRegistered as IndexNotFoundException  # noqa
         legacy = True
     except ImportError, e:
         raise ImproperlyConfigured("Haystack couldn't be imported: %s" % e)
 
+if settings.CELERY_HAYSTACK_TRANSACTION_SAFE and not settings.CELERY_ALWAYS_EAGER:
+    from djcelery_transactions import PostTransactionTask as Task
+else:
+    from celery.task import Task  # noqa
+
 
 class CeleryHaystackSignalHandler(Task):
     using = settings.CELERY_HAYSTACK_DEFAULT_ALIAS
     max_retries = settings.CELERY_HAYSTACK_MAX_RETRIES
     default_retry_delay = settings.CELERY_HAYSTACK_RETRY_DELAY
 
+    def get_logger(self, *args, **kwargs):
+        logger = super(CeleryHaystackSignalHandler, self).get_logger(*args, **kwargs)
+        if settings.DEBUG:
+            logger.setLogger(logging.DEBUG)
+        return logger
+
     def split_identifier(self, identifier, **kwargs):
         """
         Break down the identifier representing the instance.
@@ -53,39 +63,40 @@ def get_model_class(self, object_path, **kwargs):
         model_class = get_model(app_name, classname)
 
         if model_class is None:
-            logger = self.get_logger(**kwargs)
-            logger.error("Could not load model "
-                         "from '%s'. Moving on..." % object_path)
-            return None
-
+            raise ImproperlyConfigured("Could not load model '%s'." %
+                                       object_path)
         return model_class
 
     def get_instance(self, model_class, pk, **kwargs):
         """
         Fetch the instance in a standarized way.
         """
         logger = self.get_logger(**kwargs)
+        instance = None
         try:
-            instance = model_class.objects.get(pk=pk)
+            instance = model_class._default_manager.get(pk=int(pk))
         except model_class.DoesNotExist:
-            logger.error("Couldn't load model instance "
-                         "with pk #%s. Somehow it went missing?" % pk)
-            return None
+            logger.error("Couldn't load %s.%s.%s. Somehow it went missing?" %
+                         (model_class._meta.app_label.lower(),
+                          model_class._meta.object_name.lower(), pk))
         except model_class.MultipleObjectsReturned:
-            logger.error("More than one object with pk #%s. Oops?" % pk)
-            return None
-
+            logger.error("More than one object with pk %s. Oops?" % pk)
         return instance
 
     def get_index(self, model_class, **kwargs):
         """
         Fetch the model's registered ``SearchIndex`` in a standarized way.
         """
-        logger = self.get_logger(**kwargs)
         try:
+            if legacy:
+                index_holder = site
+            else:
+                backend_alias = connection_router.for_write(**{'models': [model_class]})
+                index_holder = connections[backend_alias].get_unified_index()  # noqa
             return index_holder.get_index(model_class)
         except IndexNotFoundException:
-            logger.error("Couldn't find a SearchIndex for %s." % model_class)
+            raise ImproperlyConfigured("Couldn't find a SearchIndex for %s." %
+                                       model_class)
         return None
 
     def get_handler_options(self, **kwargs):
@@ -104,58 +115,69 @@ def run(self, action, identifier, **kwargs):
         # First get the object path and pk (e.g. ('notes.note', 23))
         object_path, pk = self.split_identifier(identifier, **kwargs)
         if object_path is None or pk is None:
-            logger.error("Skipping.")
-            return
+            msg = "Couldn't handle object with identifier %s" % identifier
+            logger.error(msg)
+            raise ValueError(msg)
 
         # Then get the model class for the object path
         model_class = self.get_model_class(object_path, **kwargs)
         current_index = self.get_index(model_class, **kwargs)
+        current_index_name = ".".join([current_index.__class__.__module__,
+                                       current_index.__class__.__name__])
 
         if action == 'delete':
-            # If the object is gone, we'll use just the identifier against the
-            # index.
+            # If the object is gone, we'll use just the identifier
+            # against the index.
             try:
                 handler_options = self.get_handler_options(**kwargs)
                 current_index.remove_object(identifier, **handler_options)
             except Exception, exc:
                 logger.error(exc)
-                self.retry([action, identifier], kwargs, exc=exc)
+                self.retry(exc=exc)
             else:
-                logger.debug("Deleted '%s' from index" % identifier)
-            return
-
+                msg = ("Deleted '%s' (with %s)" %
+                       (identifier, current_index_name))
+                logger.debug(msg)
+                return msg
         elif action == 'update':
             # and the instance of the model class with the pk
             instance = self.get_instance(model_class, pk, **kwargs)
             if instance is None:
-                logger.debug("Didn't update index for '%s'" % identifier)
-                return
+                logger.debug("Failed updating '%s' (with %s)" %
+                             (identifier, current_index_name))
+                raise ValueError("Couldn't load object '%s'" % identifier)
 
             # Call the appropriate handler of the current index and
             # handle exception if neccessary
-            logger.debug("Indexing '%s'." % instance)
             try:
                 handler_options = self.get_handler_options(**kwargs)
                 current_index.update_object(instance, **handler_options)
             except Exception, exc:
                 logger.error(exc)
-                self.retry([action, identifier], kwargs, exc=exc)
+                self.retry(exc=exc)
             else:
-                logger.debug("Updated index with '%s'" % instance)
+                msg = ("Updated '%s' (with %s)" %
+                       (identifier, current_index_name))
+                logger.debug(msg)
+                return msg
         else:
             logger.error("Unrecognized action '%s'. Moving on..." % action)
-            self.retry([action, identifier], kwargs, exc=exc)
+            raise ValueError("Unrecognized action %s" % action)
 
 
 class CeleryHaystackUpdateIndex(Task):
     """
     A celery task class to be used to call the update_index management
     command from Celery.
     """
+    def get_logger(self, *args, **kwargs):
+        logger = super(CeleryHaystackUpdateIndex, self).get_logger(*args, **kwargs)
+        if settings.DEBUG:
+            logger.setLogger(logging.DEBUG)
+        return logger
+
     def run(self, apps=None, **kwargs):
         logger = self.get_logger(**kwargs)
-        logger.info("Starting update index")
-        # Run the update_index management command
         defaults = {
             'batchsize': settings.CELERY_HAYSTACK_COMMAND_BATCH_SIZE,
             'age': settings.CELERY_HAYSTACK_COMMAND_AGE,
@@ -167,5 +189,7 @@ def run(self, apps=None, **kwargs):
         defaults.update(kwargs)
         if apps is None:
             apps = settings.CELERY_HAYSTACK_COMMAND_APPS
+        # Run the update_index management command
+        logger.info("Starting update index")
         call_command('update_index', *apps, **defaults)
         logger.info("Finishing update index")

celery_haystack/test_settings.py

@@ -1,5 +1,7 @@
 import os
 
+DEBUG = True
+
 TEST_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), 'tests'))
 
 INSTALLED_APPS = [

celery_haystack/utils.py

@@ -17,4 +17,4 @@ def get_update_task(task_path=None):
     except AttributeError:
         raise ImproperlyConfigured('Module "%s" does not define a "%s" '
                                    'class.' % (module, attr))
-    return Task()
+    return Task

docs/changelog.rst

@@ -1,6 +1,30 @@
 Changelog
 =========
 
+v0.6 (2012-06-27)
+-----------------
+
+* *backwards incompatible change* Added support for
+  `django-celery-transactions`_ to make sure the tasks are respecting
+  Django's transaction management. It holds on to Celery tasks
+  until the current database transaction is committed, avoiding potential
+  race conditions as described in `Celery's user guide`_.
+
+  This is **enabled by default** but can be disabled in case you want
+  to manually manage the tranasctions:
+
+      CELERY_HAYSTACK_TRANSACTION_SAFE = False
+
+* Refactored the error handling to always return a message about what
+  happened in every step of the index interaction. Raise exception about
+  misconfiguration and wrong parameters quicker.
+
+* Improved support for multiple search indexes as implemented by
+  Haystack 2.X. Many thanks to Germán M. Bravo (Kronuz).
+
+.. _`django-celery-transactions`: https://github.com/chrisdoble/django-celery-transactions 
+.. _`Celery's user guide`: http://celery.readthedocs.org/en/latest/userguide/tasks.html#database-transactions
+
 v0.5 (2012-05-23)
 -----------------
 

setup.py

@@ -39,7 +39,8 @@ def find_version(*file_paths):
         'Topic :: Utilities',
     ],
     install_requires=[
-        'django-appconf >= 0.4.1',
+        'django-appconf>=0.4.1',
+        'django-celery-transactions>=0.1.2'
     ],
     zip_safe=False,
 )