Patch 1

django_apscheduler/jobstores.py

@@ -11,7 +11,7 @@
 from apscheduler.schedulers.base import BaseScheduler
 
 from django import db
-from django.db import transaction, IntegrityError
+from django.db import transaction, IntegrityError, router
 
 from django_apscheduler import util
 from django_apscheduler.models import DjangoJob, DjangoJobExecution
@@ -233,7 +233,8 @@ def get_all_jobs(self):
 
     @util.retry_on_db_operational_error
     def add_job(self, job: AppSchedulerJob):
-        with transaction.atomic():
+        db = router.db_for_write(DjangoJob)
+        with transaction.atomic(using=db):
             try:
                 return DjangoJob.objects.create(
                     id=job.id,
@@ -246,7 +247,8 @@ def add_job(self, job: AppSchedulerJob):
     @util.retry_on_db_operational_error
     def update_job(self, job: AppSchedulerJob):
         # Acquire lock for update
-        with transaction.atomic():
+        db = router.db_for_write(DjangoJob)
+        with transaction.atomic(using=db):
             try:
                 db_job = DjangoJob.objects.select_for_update().get(id=job.id)
 
@@ -262,7 +264,8 @@ def update_job(self, job: AppSchedulerJob):
 
     @util.retry_on_db_operational_error
     def remove_job(self, job_id: str):
-        with transaction.atomic():
+        db = router.db_for_write(DjangoJob)
+        with transaction.atomic(using=db):
             try:
                 DjangoJob.objects.select_for_update().get(id=job_id).delete()
             except DjangoJob.DoesNotExist:

django_apscheduler/migrations/0005_migrate_name_to_id.py

@@ -9,25 +9,26 @@ def migrate_name_to_id(apps, schema_editor):
     JobExecutionModel = apps.get_model("django_apscheduler", "DjangoJobExecution")
     migrated_id_mappings = {}
     migrated_job_executions = []
-
+    db = schema_editor.connection.alias
+
     # Copy 'name' to 'id'.
-    for job in JobModel.objects.all():
+    for job in JobModel.objects.using(db).all():
         migrated_id_mappings[job.id] = job.name
         job.id = job.name
         job.name = f"{job.name}_tmp"
         job.save()
 
     # Update all job execution references
-    for job_execution in JobExecutionModel.objects.filter(
+    for job_execution in JobExecutionModel.objects.using(db).filter(
         job_id__in=migrated_id_mappings
     ):
         job_execution.job_id = migrated_id_mappings[job_execution.job_id]
         migrated_job_executions.append(job_execution)
 
-    JobExecutionModel.objects.bulk_update(migrated_job_executions, ["job_id"])
+    JobExecutionModel.objects.using(db).bulk_update(migrated_job_executions, ["job_id"])
 
     # Remove old jobs
-    JobModel.objects.filter(id__in=migrated_id_mappings).delete()
+    JobModel.objects.using(db).filter(id__in=migrated_id_mappings).delete()
 
 
 class Migration(migrations.Migration):

django_apscheduler/models.py

@@ -1,6 +1,6 @@
 from datetime import timedelta, datetime
 
-from django.db import models, transaction
+from django.db import models, transaction, routers
 from django.db.models import UniqueConstraint
 from django.utils import timezone
 from django.utils.translation import gettext_lazy as _
@@ -162,7 +162,8 @@ def atomic_update_or_create(
             finished = finished.timestamp()
 
             try:
-                with transaction.atomic():
+                db = router.db_for_write(DjangoJobExecution)
+                with transaction.atomic(using=db):
                     job_execution = DjangoJobExecution.objects.select_for_update().get(
                         job_id=job_id, run_time=run_time
                     )