Horrible hacks to work with Django 1.6 and still be compatible

Author: ask

djcelery/loaders.py

@@ -72,7 +72,10 @@ def _close_database(self):
         try:
             funs = [conn.close for conn in db.connections]
         except AttributeError:
-            funs = [db.close_connection]  # pre multidb
+            if hasattr(db, 'close_old_connections'):  # Django 1.6+
+                funs = [db.close_old_connections]
+            else:
+                funs = [db.close_connection]  # pre multidb
 
         for close in funs:
             try:

djcelery/managers.py

@@ -17,6 +17,9 @@
 
 from celery.utils.timeutils import maybe_timedelta
 
+from .db import (
+    commit_on_success, commit_unless_managed, rollback_unless_managed,
+)
 from .utils import now
 
 
@@ -47,8 +50,10 @@ def _inner(*args, **kwargs):
                     # the transaction.
                     if retries >= _max_retries:
                         raise
-                    transaction.rollback_unless_managed()
-
+                    try:
+                        rollback_unless_managed()
+                    except Exception:
+                        pass
         return _inner
 
     return _outer
@@ -105,22 +110,16 @@ def get_all_expired(self, expires):
         """Get all expired task results."""
         return self.filter(date_done__lt=now() - maybe_timedelta(expires))
 
-    @transaction.commit_manually
     def delete_expired(self, expires):
         """Delete all expired taskset results."""
         meta = self.model._meta
-        try:
+        with commit_on_success():
             self.get_all_expired(expires).update(hidden=True)
             cursor = self.connection_for_write().cursor()
             cursor.execute(
                 'DELETE FROM {0.db_table} WHERE hidden=%s'.format(meta),
                 (True, ),
             )
-        except:
-            transaction.rollback()
-            raise
-        else:
-            transaction.commit()
 
 
 class PeriodicTaskManager(ExtendedManager):
@@ -244,4 +243,7 @@ def purge(self):
             'DELETE FROM {0.db_table} WHERE hidden=%s'.format(meta),
             (True, ),
         )
-        transaction.commit_unless_managed()
+        try:
+            commit_unless_managed()
+        except transaction.TransactionManagementError:
+            pass

djcelery/schedulers.py

@@ -15,6 +15,7 @@
 from django.db import transaction
 from django.core.exceptions import ObjectDoesNotExist
 
+from .db import commit_on_success
 from .models import (PeriodicTask, PeriodicTasks,
                      CrontabSchedule, IntervalSchedule)
 from .utils import DATABASE_ERRORS, make_aware
@@ -195,24 +196,18 @@ def reserve(self, entry):
         self._dirty.add(new_entry.name)
         return new_entry
 
-    @transaction.commit_manually
     def sync(self):
         info('Writing entries...')
         _tried = set()
         try:
-            try:
+            with commit_on_success():
                 while self._dirty:
                     try:
                         name = self._dirty.pop()
                         _tried.add(name)
                         self.schedule[name].save()
                     except (KeyError, ObjectDoesNotExist):
                         pass
-            except:
-                transaction.rollback()
-                raise
-            else:
-                transaction.commit()
         except DATABASE_ERRORS as exc:
             # retry later
             self._dirty |= _tried

djcelery/snapshot.py

@@ -3,7 +3,6 @@
 from collections import defaultdict
 from datetime import datetime, timedelta
 
-from django.db import transaction
 from django.conf import settings
 
 from celery import states
@@ -13,6 +12,7 @@
 from celery.utils.log import get_logger
 from celery.utils.timeutils import maybe_iso8601
 
+from .db import commit_on_success
 from .models import WorkerState, TaskState
 from .utils import maybe_make_aware
 
@@ -129,23 +129,15 @@ def update_task(self, state, **kwargs):
 
         return obj
 
-    @transaction.commit_manually
     def on_shutter(self, state, commit_every=100):
-        if not state.event_count and transaction.is_dirty():
-            transaction.commit()
-            return
 
         def _handle_tasks():
             for i, task in enumerate(state.tasks.items()):
                 self.handle_task(task)
-                if not i % commit_every:
-                    transaction.commit()
 
         for worker in state.workers.items():
             self.handle_worker(worker)
         _handle_tasks()
-        if transaction.is_dirty():
-            transaction.commit()
 
     def on_cleanup(self):
         expired = (self.TaskState.objects.expire_by_states(states, expires)

djcelery/tests/test_snapshot.py

@@ -4,7 +4,7 @@
 from itertools import count
 from time import time
 
-from celery.events import Event
+from celery.events import Event as _Event
 from celery.events.state import State, Worker, Task
 from celery.utils import gen_unique_id
 
@@ -15,6 +15,12 @@
 from djcelery.tests.utils import unittest
 
 _next_id = count(0).next
+_next_clock = count(1).next
+
+def Event(*args, **kwargs):
+    kwargs.setdefault('clock', _next_clock())
+    kwargs.setdefault('local_received', time())
+    return _Event(*args, **kwargs)
 
 
 def create_task(worker, **kwargs):