Optional Cron expression scheduling - inital commit

django_q/admin.py

@@ -10,14 +10,7 @@ from django_q.tasks import async_task
 class TaskAdmin(admin.ModelAdmin):
     """model admin for success tasks."""
 
-    list_display = (
-        'name',
-        'func',
-        'started',
-        'stopped',
-        'time_taken',
-        'group'
-    )
+    list_display = ("name", "func", "started", "stopped", "time_taken", "group")
 
     def has_add_permission(self, request):
         """Don't allow adds."""
@@ -28,9 +21,9 @@ class TaskAdmin(admin.ModelAdmin):
         qs = super(TaskAdmin, self).get_queryset(request)
         return qs.filter(success=True)
 
-    search_fields = ('name', 'func', 'group')
+    search_fields = ("name", "func", "group")
     readonly_fields = []
-    list_filter = ('group',)
+    list_filter = ("group",)
 
     def get_readonly_fields(self, request, obj=None):
         """Set all fields readonly."""
@@ -50,21 +43,15 @@ retry_failed.short_description = _("Resubmit selected tasks to queue")
 class FailAdmin(admin.ModelAdmin):
     """model admin for failed tasks."""
 
-    list_display = (
-        'name',
-        'func',
-        'started',
-        'stopped',
-        'short_result'
-    )
+    list_display = ("name", "func", "started", "stopped", "short_result")
 
     def has_add_permission(self, request):
         """Don't allow adds."""
         return False
 
     actions = [retry_failed]
-    search_fields = ('name', 'func')
-    list_filter = ('group',)
+    search_fields = ("name", "func")
+    list_filter = ("group",)
     readonly_fields = []
 
     def get_readonly_fields(self, request, obj=None):
@@ -76,31 +63,31 @@ class ScheduleAdmin(admin.ModelAdmin):
     """ model admin for schedules """
 
     list_display = (
-        'id',
-        'name',
-        'func',
-        'schedule_type',
-        'repeats',
-        'next_run',
-        'last_run',
-        'success'
+        "id",
+        "name",
+        "func",
+        "schedule_type",
+        "repeats",
+        "next_run",
+        "last_run",
+        "success",
     )
 
-    list_filter = ('next_run', 'schedule_type')
-    search_fields = ('func',)
-    list_display_links = ('id', 'name')
+    # optional cron strings
+    try:
+        from croniter import croniter
+    except ImportError:
+        readonly_fields = ("cron",)
+
+    list_filter = ("next_run", "schedule_type")
+    search_fields = ("func",)
+    list_display_links = ("id", "name")
 
 
 class QueueAdmin(admin.ModelAdmin):
     """  queue admin for ORM broker """
-    list_display = (
-        'id',
-        'key',
-        'task_id',
-        'name',
-        'func',
-        'lock'
-    )
+
+    list_display = ("id", "key", "task_id", "name", "func", "lock")
 
     def save_model(self, request, obj, form, change):
         obj.save(using=Conf.ORM)

django_q/cluster.py

@@ -1,6 +1,5 @@
-import ast
-
 # Standard
+import ast
 import importlib
 import signal
 import socket
@@ -9,9 +8,8 @@ import uuid
 from multiprocessing import Event, Process, Value, current_process
 from time import sleep
 
-# external
+# External
 import arrow
-
 # Django
 from django import db
 from django.conf import settings
@@ -29,6 +27,12 @@ from django_q.signals import pre_execute
 from django_q.signing import SignedPackage, BadSignature
 from django_q.status import Stat, Status
 
+# Optional
+try:
+    from croniter import croniter
+except ImportError:
+    croniter = None
+
 
 class Cluster:
     def __init__(self, broker: Broker = None):
@@ -103,10 +107,10 @@ class Cluster:
     @property
     def is_stopping(self) -> bool:
         return (
-            self.stop_event
-            and self.start_event
-            and self.start_event.is_set()
-            and self.stop_event.is_set()
+                self.stop_event
+                and self.start_event
+                and self.start_event.is_set()
+                and self.stop_event.is_set()
         )
 
     @property
@@ -116,13 +120,13 @@ class Cluster:
 
 class Sentinel:
     def __init__(
-        self,
-        stop_event,
-        start_event,
-        cluster_id,
-        broker=None,
-        timeout=Conf.TIMEOUT,
-        start=True,
+            self,
+            stop_event,
+            start_event,
+            cluster_id,
+            broker=None,
+            timeout=Conf.TIMEOUT,
+            start=True,
     ):
         # Make sure we catch signals for the pool
         signal.signal(signal.SIGINT, signal.SIG_IGN)
@@ -377,7 +381,7 @@ def monitor(result_queue: Queue, broker: Broker = None):
 
 
 def worker(
-    task_queue: Queue, result_queue: Queue, timer: Value, timeout: int = Conf.TIMEOUT
+        task_queue: Queue, result_queue: Queue, timer: Value, timeout: int = Conf.TIMEOUT
 ):
     """
     Takes a task from the task queue, tries to execute it and puts the result back in the result queue
@@ -552,9 +556,9 @@ def scheduler(broker: Broker = None):
     try:
         with db.transaction.atomic(using=Schedule.objects.db):
             for s in (
-                Schedule.objects.select_for_update()
-                .exclude(repeats=0)
-                .filter(next_run__lt=timezone.now())
+                    Schedule.objects.select_for_update()
+                            .exclude(repeats=0)
+                            .filter(next_run__lt=timezone.now())
             ):
                 args = ()
                 kwargs = {}
@@ -591,6 +595,16 @@ def scheduler(broker: Broker = None):
                             next_run = next_run.shift(months=+3)
                         elif s.schedule_type == s.YEARLY:
                             next_run = next_run.shift(years=+1)
+                        elif s.schedule_type == s.CRON:
+                            if not croniter:
+                                raise ImportError(
+                                    _(
+                                        "Please install croniter to enable cron expressions"
+                                    )
+                                )
+                            next_run = arrow.get(
+                                croniter(s.cron, timezone.now()).get_next()
+                            )
                         if Conf.CATCH_UP or next_run > arrow.utcnow():
                             break
                     # arrow always returns a tz aware datetime, and we don't want

django_q/models.py

@@ -1,16 +1,26 @@
+# Django
 from django import get_version
+from django.core.exceptions import ValidationError
+from django.db import models
 from django.template.defaultfilters import truncatechars
-
 from django.urls import reverse
+from django.utils import timezone
 from django.utils.html import format_html
 from django.utils.translation import gettext_lazy as _
-from django.db import models
-from django.utils import timezone
+
+# External
 from picklefield import PickledObjectField
 from picklefield.fields import dbsafe_decode
 
+# Local
 from django_q.signing import SignedPackage
 
+# Optional
+try:
+    from croniter import croniter
+except ImportError:
+    croniter = None
+
 
 class Task(models.Model):
     id = models.CharField(max_length=32, primary_key=True, editable=False)
@@ -131,6 +141,18 @@ class Failure(Task):
         proxy = True
 
 
+# Optional Cron validator
+def validate_cron(value):
+    if not value:
+        return
+    if not croniter:
+        raise ImportError(_("Please install croniter to enable cron expressions"))
+    try:
+        croniter.expand(value)
+    except ValueError as e:
+        raise ValidationError(e)
+
+
 class Schedule(models.Model):
     name = models.CharField(max_length=100, null=True, blank=True)
     func = models.CharField(max_length=256, help_text="e.g. module.tasks.function")
@@ -152,6 +174,7 @@ class Schedule(models.Model):
     MONTHLY = "M"
     QUARTERLY = "Q"
     YEARLY = "Y"
+    CRON = "C"
     TYPE = (
         (ONCE, _("Once")),
         (MINUTES, _("Minutes")),
@@ -161,6 +184,7 @@ class Schedule(models.Model):
         (MONTHLY, _("Monthly")),
         (QUARTERLY, _("Quarterly")),
         (YEARLY, _("Yearly")),
+        (CRON, _("Cron")),
     )
     schedule_type = models.CharField(
         max_length=1, choices=TYPE, default=TYPE[0][0], verbose_name=_("Schedule Type")
@@ -174,6 +198,13 @@ class Schedule(models.Model):
     next_run = models.DateTimeField(
         verbose_name=_("Next Run"), default=timezone.now, null=True
     )
+    cron = models.CharField(
+        max_length=100,
+        null=True,
+        blank=True,
+        validators=[validate_cron],
+        help_text=_("Cron expression"),
+    )
     task = models.CharField(max_length=100, null=True, editable=False)
 
     def success(self):