feat(schedulers): add StrictDatabaseScheduler that disables unknown tasks

Adds a new DatabaseScheduler subclass that, at startup, disables any
enabled PeriodicTask whose dotted task name is not registered with the
running Celery app. This addresses the common operational pain point
where tasks are removed from the codebase but their database rows
linger with enabled=True, causing beat to keep dispatching them and
workers to log NotRegistered errors indefinitely.

The new scheduler is exposed both as a class
(django_celery_beat.schedulers:StrictDatabaseScheduler) and as a
celery beat-scheduler entry point alias (django_strict). Disabled
rows are not deleted; they can be re-enabled from the admin if the
task name becomes registered again.

Author: serl

django_celery_beat/schedulers.py

@@ -542,3 +542,30 @@ def schedule(self):
                     repr(entry) for entry in self._schedule.values()),
                 )
         return self._schedule
+
+
+class StrictDatabaseScheduler(DatabaseScheduler):
+    """A :class:`DatabaseScheduler` that disables any periodic task whose
+    Celery task name is not registered with the running app.
+
+    Useful in deployments where periodic tasks may be removed
+    from the codebase while their database rows linger with
+    ``enabled=True``. With the default :class:`DatabaseScheduler`, beat
+    keeps dispatching those tasks and workers raise ``NotRegistered``
+    indefinitely. With this scheduler, such tasks are disabled at
+    startup so they stop being dispatched.
+    """
+
+    def setup_schedule(self):
+        super().setup_schedule()
+        self._disable_unknown_tasks()
+
+    def _disable_unknown_tasks(self):
+        for task in self.Model.objects.enabled():
+            if task.task not in self.app.tasks:
+                warning(
+                    'Disabling unregistered periodic task %r (task=%r)',
+                    task.name, task.task,
+                )
+                task.enabled = False
+                task.save()

docs/includes/introduction.txt

@@ -228,6 +228,31 @@ with only one command (recommended for **development environment only**):
 3. Now you can add and manage your periodic tasks from the Django Admin interface.
 
 
+Choosing a scheduler
+~~~~~~~~~~~~~~~~~~~~
+
+This package ships two scheduler classes. Pick based on whether the
+beat process has the same task registry as the workers that consume
+its messages:
+
+- ``django_celery_beat.schedulers:StrictDatabaseScheduler``
+  (alias ``django_strict``): **recommended when all periodic tasks are
+  defined in the same Django app that runs beat** — i.e. beat imports
+  every task it dispatches. At startup it disables any enabled
+  ``PeriodicTask`` whose dotted task name is not registered with the
+  running Celery app, which prevents ``NotRegistered`` errors that
+  otherwise persist indefinitely after a task is removed from the
+  codebase. Disabled rows are not deleted; they can be re-enabled from
+  the admin if the task name becomes registered again.
+
+- ``django_celery_beat.schedulers:DatabaseScheduler`` (alias
+  ``django``): **the default — use it for mixed deployments** where
+  beat dispatches tasks to workers running different codebases (and
+  therefore the beat process does not import every task it routes).
+  In that setup, an unknown name is expected, not an error, so the
+  strict variant would incorrectly disable valid tasks.
+
+
 
 Working with django-celery-results
 -----------------------------------

setup.py

@@ -126,6 +126,7 @@ def reqs(*f):
     entry_points={
         'celery.beat_schedulers': [
             'django = django_celery_beat.schedulers:DatabaseScheduler',
+            'django_strict = django_celery_beat.schedulers:StrictDatabaseScheduler',
         ],
     },
     include_package_data=True,

t/unit/test_schedulers.py

@@ -1408,6 +1408,37 @@ def test_scheduler_valid_hours(self):
             assert 0 <= hour_value <= 23
 
 
+@pytest.mark.django_db
+class test_StrictDatabaseScheduler(SchedulerCase):
+    Scheduler = schedulers.StrictDatabaseScheduler
+
+    @pytest.fixture(autouse=True)
+    def setup_scheduler(self, app):
+        self.app = app
+        self.app.conf.beat_schedule = {}
+
+        self.known = self.create_model_interval(
+            schedule(timedelta(seconds=10)),
+            task='celery.backend_cleanup',
+        )
+        self.known.save()
+
+        self.unknown = self.create_model_interval(
+            schedule(timedelta(seconds=10)),
+            task='nonexistent.task.that.was.removed',
+        )
+        self.unknown.save()
+
+    def test_unregistered_task_is_disabled(self):
+        self.Scheduler(app=self.app)
+
+        self.unknown.refresh_from_db()
+        assert self.unknown.enabled is False
+
+        self.known.refresh_from_db()
+        assert self.known.enabled is True
+
+
 @pytest.mark.django_db
 class test_models(SchedulerCase):