Abstract Django DB Models

.gitignore

@@ -34,3 +34,4 @@ coverage.xml
 .python-version
 venv
 .env
+.vscode/

README.rst

@@ -86,6 +86,54 @@ manually:
         >>> from django_celery_beat.models import PeriodicTasks
         >>> PeriodicTasks.update_changed()
 
+Custom Models
+=============
+
+It's possible to use your own models instead of the default ones provided by ``django_celery_beat``, to do that just define your models inheriting from the right one from ``django_celery_beat.models.abstract``:
+
+.. code-block:: Python
+
+        # custom_app.models.py
+        from django_celery_beat.models.abstract import (
+            AbstractClockedSchedule,
+            AbstractCrontabSchedule,
+            AbstractIntervalSchedule,
+            AbstractPeriodicTask,
+            AbstractPeriodicTasks,
+            AbstractSolarSchedule,
+        )
+
+        class CustomPeriodicTask(AbstractPeriodicTask):
+            ...
+
+        class CustomPeriodicTasks(AbstractPeriodicTasks):
+            ...
+
+        class CustomCrontabSchedule(AbstractCrontabSchedule):
+            ...
+
+        class CustomIntervalSchedule(AbstractIntervalSchedule):
+            ...
+
+        class CustomSolarSchedule(AbstractSolarSchedule):
+            ...
+
+        class CustomClockedSchedule(AbstractClockedSchedule):
+            ...
+
+To let ``django_celery_beat.scheduler`` make use of your own modules, you must provide the ``app_name.model_name`` of your own custom models as values to the next constants in your settings:
+
+.. code-block:: Python
+
+        # settings.py
+        # CELERY_BEAT_(?:PERIODICTASKS?|(?:CRONTAB|INTERVAL|SOLAR|CLOCKED)SCHEDULE)_MODEL = "app_label.model_name"
+        CELERY_BEAT_PERIODICTASK_MODEL = "custom_app.CustomPeriodicTask"
+        CELERY_BEAT_PERIODICTASKS_MODEL = "custom_app.CustomPeriodicTasks"
+        CELERY_BEAT_CRONTABSCHEDULE_MODEL = "custom_app.CustomCrontabSchedule"
+        CELERY_BEAT_INTERVALSCHEDULE_MODEL = "custom_app.CustomIntervalSchedule"
+        CELERY_BEAT_SOLARSCHEDULE_MODEL = "custom_app.CustomSolarSchedule"
+        CELERY_BEAT_CLOCKEDSCHEDULE_MODEL = "custom_app.CustomClockedSchedule"
+
 Example creating interval-based periodic task
 ---------------------------------------------
 

django_celery_beat/helpers.py

@@ -0,0 +1,135 @@
+from django.apps import apps
+from django.conf import settings
+from django.core.exceptions import ImproperlyConfigured
+
+
+def crontabschedule_model():
+    """Return the CrontabSchedule model that is active in this project."""
+    if not hasattr(settings, "CELERY_BEAT_CRONTABSCHEDULE_MODEL"):
+        from .models.generic import CrontabSchedule
+
+        return CrontabSchedule
+
+    try:
+        return apps.get_model(settings.CELERY_BEAT_CRONTABSCHEDULE_MODEL)
+    except ValueError:
+        raise ImproperlyConfigured(
+            "CELERY_BEAT_CRONTABSCHEDULE_MODEL must be of the form "
+            "'app_label.model_name'"
+        )
+    except LookupError:
+        raise ImproperlyConfigured(
+            "CELERY_BEAT_CRONTABSCHEDULE_MODEL refers to model "
+            f"'{settings.CELERY_BEAT_CRONTABSCHEDULE_MODEL}' that has not "
+            "been installed"
+        )
+
+
+def intervalschedule_model():
+    """Return the IntervalSchedule model that is active in this project."""
+    if not hasattr(settings, "CELERY_BEAT_INTERVALSCHEDULE_MODEL"):
+        from .models.generic import IntervalSchedule
+
+        return IntervalSchedule
+
+    try:
+        return apps.get_model(settings.CELERY_BEAT_INTERVALSCHEDULE_MODEL)
+    except ValueError:
+        raise ImproperlyConfigured(
+            "CELERY_BEAT_INTERVALSCHEDULE_MODEL must be of the form "
+            "'app_label.model_name'"
+        )
+    except LookupError:
+        raise ImproperlyConfigured(
+            "CELERY_BEAT_INTERVALSCHEDULE_MODEL refers to model "
+            f"'{settings.CELERY_BEAT_INTERVALSCHEDULE_MODEL}' that has not "
+            "been installed"
+        )
+
+
+def periodictask_model():
+    """Return the PeriodicTask model that is active in this project."""
+    if not hasattr(settings, "CELERY_BEAT_PERIODICTASK_MODEL"):
+        from .models.generic import PeriodicTask
+
+        return PeriodicTask
+
+    try:
+        return apps.get_model(settings.CELERY_BEAT_PERIODICTASK_MODEL)
+    except ValueError:
+        raise ImproperlyConfigured(
+            "CELERY_BEAT_PERIODICTASK_MODEL must be of the form "
+            "'app_label.model_name'"
+        )
+    except LookupError:
+        raise ImproperlyConfigured(
+            "CELERY_BEAT_PERIODICTASK_MODEL refers to model "
+            f"'{settings.CELERY_BEAT_PERIODICTASK_MODEL}' that has not been "
+            "installed"
+        )
+
+
+def periodictasks_model():
+    """Return the PeriodicTasks model that is active in this project."""
+    if not hasattr(settings, "CELERY_BEAT_PERIODICTASKS_MODEL"):
+        from .models.generic import PeriodicTasks
+
+        return PeriodicTasks
+
+    try:
+        return apps.get_model(settings.CELERY_BEAT_PERIODICTASKS_MODEL)
+    except ValueError:
+        raise ImproperlyConfigured(
+            "CELERY_BEAT_PERIODICTASKS_MODEL must be of the form "
+            "'app_label.model_name'"
+        )
+    except LookupError:
+        raise ImproperlyConfigured(
+            "CELERY_BEAT_PERIODICTASKS_MODEL refers to model "
+            f"'{settings.CELERY_BEAT_PERIODICTASKS_MODEL}' that has not been "
+            "installed"
+        )
+
+
+def solarschedule_model():
+    """Return the SolarSchedule model that is active in this project."""
+    if not hasattr(settings, "CELERY_BEAT_SOLARSCHEDULE_MODEL"):
+        from .models.generic import SolarSchedule
+
+        return SolarSchedule
+
+    try:
+        return apps.get_model(settings.CELERY_BEAT_SOLARSCHEDULE_MODEL)
+    except ValueError:
+        raise ImproperlyConfigured(
+            "CELERY_BEAT_SOLARSCHEDULE_MODEL must be of the form "
+            "'app_label.model_name'"
+        )
+    except LookupError:
+        raise ImproperlyConfigured(
+            "CELERY_BEAT_SOLARSCHEDULE_MODEL refers to model "
+            f"'{settings.CELERY_BEAT_SOLARSCHEDULE_MODEL}' that has not been "
+            "installed"
+        )
+
+
+def clockedschedule_model():
+    """Return the ClockedSchedule model that is active in this project."""
+    if not hasattr(settings, "CELERY_BEAT_CLOCKEDSCHEDULE_MODEL"):
+        from .models.generic import ClockedSchedule
+
+        return ClockedSchedule
+
+    try:
+        return apps.get_model(settings.CELERY_BEAT_CLOCKEDSCHEDULE_MODEL)
+    except ValueError:
+        raise ImproperlyConfigured(
+            "CELERY_BEAT_CLOCKEDSCHEDULE_MODEL must be of the form "
+            "'app_label.model_name'"
+        )
+    except LookupError:
+        raise ImproperlyConfigured(
+            "CELERY_BEAT_CLOCKEDSCHEDULE_MODEL refers to model "
+            f"'{settings.CELERY_BEAT_CLOCKEDSCHEDULE_MODEL}' that has not "
+            "been installed"
+        )

django_celery_beat/models/__init__.py

@@ -0,0 +1,14 @@
+from .abstract import DAYS, crontab_schedule_celery_timezone
+from .generic import (ClockedSchedule, CrontabSchedule, IntervalSchedule,
+                      PeriodicTask, PeriodicTasks, SolarSchedule)
+
+__all__ = [
+    "ClockedSchedule",
+    "CrontabSchedule",
+    "IntervalSchedule",
+    "PeriodicTask",
+    "PeriodicTasks",
+    "SolarSchedule",
+    "crontab_schedule_celery_timezone",
+    "DAYS",
+]