Add persistent schedule using redis jobstore

README.md

@@ -38,10 +38,15 @@ Finally, you lauch the scheduler in a separate process:
 python3 manage.py crontab
 ```
 
-### Setup Redis as a lock backend (optional)
+### Setup Redis as a lock backend and persistent schedule (optional)
 
 If you use Redis as a broker, you can use Redis as a lock backend as well.
-The lock backend is used to prevent multiple instances of the scheduler from running at the same time.
+The lock backend is used to prevent multiple instances of the scheduler
+from running at the same time. This is important if you have multiple
+instances of your application running.
+
+You can also use Redis as a persistent schedule backend. This is useful
+if you want to keep the schedule between restarts of the scheduler.
 
 ```python
 # settings.py
@@ -88,4 +93,4 @@ options:
 
 [dramatiq]: https://dramatiq.io/
 [apscheduler]: https://apscheduler.readthedocs.io/en/stable/
-[sentry]: https://docs.sentry.io/product/crons/
+[sentry]: https://docs.sentry.io/product/crons/

dramatiq_crontab/__init__.py

@@ -5,6 +5,8 @@
 from django.utils import timezone
 
 from . import _version
+from .conf import get_settings
+from .utils import jobstores
 
 try:
     from sentry_sdk.crons import monitor
@@ -18,7 +20,7 @@
 __all__ = ["cron", "scheduler"]
 
 
-scheduler = BlockingScheduler()
+scheduler = BlockingScheduler(jobstores=jobstores)
 
 
 def cron(schedule):

dramatiq_crontab/management/commands/crontab.py

@@ -4,33 +4,21 @@
 from django.apps import apps
 from django.core.management import BaseCommand
 
-try:
-    import redis
-    from redis.exceptions import LockError
-except ImportError:
-    redis = None
-    LockError = Exception
+from ...utils import LockError, lock
 
 try:
     from sentry_sdk import capture_exception
 except ImportError:
     capture_exception = lambda e: None
 
 from ... import scheduler
-from ...conf import get_settings
 
 
 def kill_softly(signum, frame):
     """Raise a KeyboardInterrupt to stop the scheduler and release the lock."""
     raise KeyboardInterrupt("Received SIGTERM!")
 
 
-def get_redis_client():
-    redis_url = get_settings().REDIS_URL
-    if redis and redis_url:
-        return redis.Redis.from_url(redis_url)
-
-
 class Command(BaseCommand):
     """Run dramatiq task scheduler for all tasks with the `cron` decorator."""
 
@@ -54,17 +42,13 @@ def handle(self, *args, **options):
         if not options["no_heartbeat"]:
             importlib.import_module("dramatiq_crontab.tasks")
             self.stdout.write("Scheduling heartbeat.")
-        redis = get_redis_client()
-        if redis:
-            try:
-                # Lock scheduler to prevent multiple instances from running.
-                with redis.lock("dramatiq-scheduler", blocking_timeout=0):
-                    self.launch_scheduler()
-            except LockError as e:
-                capture_exception(e)
-                self.stderr.write("Another scheduler is already running.")
-        else:
-            self.launch_scheduler()
+        try:
+            # Lock scheduler to prevent multiple instances from running.
+            with lock:
+                self.launch_scheduler()
+        except LockError as e:
+            capture_exception(e)
+            self.stderr.write("Another scheduler is already running.")
 
     def launch_scheduler(self):
         signal.signal(signal.SIGTERM, kill_softly)

dramatiq_crontab/utils.py

@@ -0,0 +1,30 @@
+from contextlib import contextmanager
+
+from dramatiq_crontab.conf import get_settings
+
+if redis_url := get_settings().REDIS_URL:
+    import redis
+    from apscheduler.jobstores.redis import RedisJobStore
+    from redis.exceptions import LockError  # noqa
+
+    redis_client = redis.Redis.from_url(redis_url)
+    lock = redis_client.lock("dramatiq-scheduler", blocking_timeout=0)
+
+    jobstores = {
+        "default": RedisJobStore(
+            connection_pool=redis.ConnectionPool.from_url(get_settings().REDIS_URL)
+        )
+    }
+
+
+else:
+
+    @contextmanager
+    def lock():
+        """Dummy lock context manager."""
+        yield
+
+    LockError = Exception
+    jobstores = {}
+
+__all__ = ["LockError", "lock", "jobstores"]

tests/test_commands.py

@@ -4,6 +4,7 @@
 import pytest
 from django.core.management import call_command
 
+import dramatiq_crontab.utils
 from dramatiq_crontab.management.commands import crontab
 
 
@@ -12,16 +13,6 @@ def test_kill_softly():
         crontab.kill_softly(None, None)
 
 
-def test_get_redis_client__none(settings):
-    settings.DRAMATIQ_CRONTAB = {}
-    assert crontab.get_redis_client() is None
-
-
-@pytest.mark.skipif(crontab.redis is None, reason="redis is not installed")
-def test_get_redis_client():
-    assert crontab.get_redis_client() is not None
-
-
 class TestCrontab:
     @pytest.fixture()
     def patch_launch(self, monkeypatch):
@@ -48,9 +39,9 @@ def test_no_heartbeat(self, patch_launch):
             assert "Loaded tasks from tests.testapp." in stdout.getvalue()
             assert "Scheduling heartbeat." not in stdout.getvalue()
 
-    @pytest.mark.skipif(crontab.redis is None, reason="redis is not installed")
     def test_locked(self):
-        with crontab.get_redis_client().lock("dramatiq-scheduler", blocking_timeout=0):
+        pytest.importorskip("redis", reason="redis is not installed")
+        with dramatiq_crontab.utils.lock:
             with io.StringIO() as stderr:
                 call_command("crontab", stderr=stderr)
                 assert "Another scheduler is already running." in stderr.getvalue()

tests/testapp/settings.py

@@ -85,7 +85,7 @@
     }
 }
 
-DRAMATIQ_CRONTAB = {"REDIS_URL": os.getenv("REDIS_URL", "redis://localhost:6379/0")}
+DRAMATIQ_CRONTAB = {"REDIS_URL": os.getenv("REDIS_URL")}
 
 dramatiq.set_broker(StubBroker())