Fix atomic transaction not routing to the the correct DB

django_celery_results/managers.py

@@ -87,7 +87,7 @@ def get_all_expired(self, expires):
 
     def delete_expired(self, expires):
         """Delete all expired results."""
-        with transaction.atomic():
+        with transaction.atomic(using=self.db):
             raw_delete(queryset=self.get_all_expired(expires))
 
 

t/proj/db_routers.py

@@ -0,0 +1,26 @@
+class AlwaysSecondaryDbRouter:
+    connection_name = "secondary"
+
+    def db_for_read(self, model, **hints):
+        """
+        Route read always for the specified connection
+        """
+        return self.connection_name
+
+    def db_for_write(self, model, **hints):
+        """
+        Route write always for the specified connection
+        """
+        return self.connection_name
+
+    def allow_relation(self, obj1, obj2, **hints):
+        """
+        Router have no opinion here
+        """
+        return None
+
+    def allow_migrate(self, db, app_label, model_name=None, **hints):
+        """
+        Router have no opinion here
+        """
+        return None

t/unit/test_models.py

@@ -5,7 +5,7 @@
 from celery import states, uuid
 from django.db import transaction
 from django.db.utils import InterfaceError
-from django.test import TransactionTestCase
+from django.test import TransactionTestCase, override_settings
 
 from django_celery_results.backends import DatabaseBackend
 from django_celery_results.models import GroupResult, TaskResult
@@ -209,3 +209,21 @@ class TransactionError(Exception):
                 raise TransactionError()
         except TransactionError:
             pass
+
+
+@pytest.mark.usefixtures('depends_on_current_app')
+class test_ModelsOnSecondaryDbOnly(TransactionTestCase):
+    """
+    These tests will fail with the below error incase we
+    try to read/write from a db other than the secondary
+
+    AssertionError: Database connections to 'default' are
+    not allowed in this test.
+    """
+    databases = ['secondary']
+
+    @override_settings(
+        DATABASE_ROUTERS=['t.proj.db_routers.AlwaysSecondaryDbRouter'],
+    )
+    def test_operations_with_atomic_transactions(self):
+        TaskResult.objects.delete_expired(expires=10)