Add schedule task func

django_scheduled_tasks/migrations/0005_scheduledtask_day_scheduledtask_hour_and_more.py

@@ -0,0 +1,29 @@
+# Generated by Django 5.0.7 on 2024-08-23 13:38
+
+import django.core.validators
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('django_scheduled_tasks', '0004_scheduledtask_exclusive'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='scheduledtask',
+            name='day',
+            field=models.CharField(blank=True, max_length=3, null=True),
+        ),
+        migrations.AddField(
+            model_name='scheduledtask',
+            name='hour',
+            field=models.SmallIntegerField(blank=True, null=True, validators=[django.core.validators.MinValueValidator(0), django.core.validators.MaxValueValidator(23)]),
+        ),
+        migrations.AlterField(
+            model_name='scheduledtask',
+            name='interval_minutes',
+            field=models.IntegerField(blank=True, null=True, validators=[django.core.validators.MinValueValidator(1), django.core.validators.MaxValueValidator(1440)]),
+        ),
+    ]

django_scheduled_tasks/models.py

@@ -15,14 +15,25 @@ class ScheduledTask(models.Model):
     """A record represents a single scheduled (recurring) task."""
 
     func = models.CharField(max_length=100, unique=True)  # importable function name
-    interval_minutes = models.IntegerField(validators=[MinValueValidator(1),
-                                                       MaxValueValidator(1440)])  # 1 min to 24 hours
+
     onstart = models.BooleanField(default=False)  # run at django startup?
     enabled = models.BooleanField(default=True)
     exclusive = models.BooleanField(default=True)  # run as the only task (I.e. thread locked)
+
     last_timestamp = models.DateTimeField(null=True, blank=True)
     last_success = models.BooleanField(null=True, blank=True)
     last_runtime = models.FloatField(null=True, blank=True)
+    day = models.CharField(max_length=3, null=True, blank=True)  # Day of the week ('mon', 'tue', etc.)
+    hour = models.SmallIntegerField(
+        validators=[MinValueValidator(0), MaxValueValidator(23)],
+        null=True,
+        blank=True
+    )  # Hour of the day (24-hour format)
+    interval_minutes = models.IntegerField(
+        validators=[MinValueValidator(1), MaxValueValidator(1440)],
+        null=True,
+        blank=True
+    )  # 1 min to 24 hours
 
     _exclusive_lock = Lock()
 

django_scheduled_tasks/register.py

@@ -15,6 +15,9 @@
 logger = logging.getLogger(__name__)
 
 
+DEFAULT_SCHEDULE_HOUR = 2
+
+
 def register_task(interval, onstart=False):
     """
     Register a scheduled task via this decorator.
@@ -45,3 +48,30 @@ def wrapper(func):
 
         return func
     return wrapper
+
+
+def schedule_task(day, hour=DEFAULT_SCHEDULE_HOUR, onstart=False):
+    """Add a scheduled task for a specific day of the week.
+
+    Args:
+         day (str): Day of the week shorthand 'mon', 'tue', 'wed', etc.
+         hour (int, optional): Hour of the day, 24-hour clock, defaults to `DEFAULT_SCHEDULE_HOUR`.
+         onstart (bool, optional): Should this be run at startup, defaults to `False`.
+    """
+    def wrapper(func):
+        try:
+            desc = f'{func.__module__}.{func.__name__}'
+            ScheduledTask.objects.create(
+                func=desc,
+                day=day,
+                hour=hour,
+                onstart=onstart
+            )
+        except IntegrityError:
+            # Already registered
+            pass
+        else:
+            logging.info(f"Registered scheduled task {desc} for day {day} at hour {hour}.")
+
+        return func
+    return wrapper

django_scheduled_tasks/scheduler.py

@@ -8,6 +8,7 @@
 import logging
 import os
 from apscheduler.schedulers.background import BackgroundScheduler
+from apscheduler.triggers.cron import CronTrigger
 from .models import ScheduledTask
 
 logger = logging.getLogger(__name__)
@@ -40,6 +41,11 @@ def add_task(func, minutes, next_run_time):
     _scheduler.add_job(func, 'interval', minutes=minutes, next_run_time=next_run_time)
 
 
+def add_day_task(func, day, hour, next_run_time):
+    """Add a task to our scheduler. Call function 'func' on every 'day' at 'hour' hours."""
+    _scheduler.add_job(func, trigger=CronTrigger(day_of_week=day, hour=hour), next_run_time=next_run_time)
+
+
 def _load_tasks():
     """Load our tasks from the model and add them to the scheduler."""
     tasks = ScheduledTask.objects.filter(enabled=True)
@@ -49,5 +55,9 @@ def _load_tasks():
     soon = datetime.now() + timedelta(minutes=1)
 
     for task in tasks:
-        add_task(task.execute, task.interval_minutes, next_run_time=soon if task.onstart else None)
+        if task.interval_minutes:
+            add_task(task.execute, task.interval_minutes, next_run_time=soon if task.onstart else None)
+        else:
+            add_day_task(task.execute, task.day, task.hour, next_run_time=soon if task.onstart else None)
+
         logging.info(f"Added task '{task}' to background scheduler")

django_scheduled_tasks/tests/test_models.py

@@ -0,0 +1,35 @@
+"""Test for django_scheduled_tasks.models."""
+from django.test import TestCase
+from django_scheduled_tasks.models import ScheduledTask
+from unittest.mock import patch
+
+
+def test_func():
+    """Dummy test function."""
+    pass
+
+
+class ScheduledTaskCase(TestCase):
+    """Tests for ScheduledTask model."""
+
+    @classmethod
+    def setUpTestData(cls):
+        """Set up test data."""
+        cls.t1 = ScheduledTask.objects.create(
+            func='django_scheduled_tasks.tests.test_models.test_func',
+            interval_minutes=10
+        )
+
+    def test_str(self):
+        """Test the str dunder method."""
+        self.assertEqual(
+            str(self.t1),
+            f"Run {self.t1.func} every {self.t1.interval_minutes} minutes"
+            f" ({'enabled' if self.t1.enabled else 'disabled'})"
+        )
+
+    @patch('django_scheduled_tasks.tests.test_models.test_func')
+    def test_execute(self, mock_test_func):
+        """Test that the execute method runs an instance of the dummy function."""
+        self.t1.execute()
+        mock_test_func.assert_called_once()

django_scheduled_tasks/tests/test_register.py

@@ -0,0 +1,59 @@
+"""Example test case."""
+from django_scheduled_tasks.models import ScheduledTask
+from django_scheduled_tasks.register import register_task, schedule_task
+from django.test import TestCase
+from unittest.mock import patch
+
+
+class RegisterTaskTest(TestCase):
+    """Test case for the `register_task` wrapper."""
+
+    def test_register_task(self):
+        """Checks that the register is empty, and then creates an object."""
+        self.assertFalse(ScheduledTask.objects.all().exists())
+
+        @register_task(1)
+        def dummy():
+            pass
+
+        self.assertEqual(ScheduledTask.objects.count(), 1)
+        st = ScheduledTask.objects.first()
+        self.assertEqual(st.func, "django_scheduled_tasks.tests.test_register.dummy")
+
+    @patch('django_scheduled_tasks.register.settings', DISABLE_SCHEDULED_TASKS=True)
+    def test_register_task_with_disable_register_task(self, mock_settings):
+        """Tests that disabling in settings correctly stops a task form being registered."""
+        @register_task(1)
+        def dummy():
+            pass
+
+        self.assertFalse(ScheduledTask.objects.all().exists())
+
+    @patch('django_scheduled_tasks.register.sys', argv=['test'])
+    def test_resgister_task_with_skipped_args(self, mock_sys_arg):
+        """Tests that running with `test` argument stops a task form being registered."""
+        @register_task(1)
+        def dummy():
+            pass
+
+        self.assertFalse(ScheduledTask.objects.all().exists())
+
+
+class ScheduledTaskTestCase(TestCase):
+    """Testing the schedule_task function."""
+
+    def test_schedule_task(self):
+        """Test decorator registers task."""
+        @schedule_task('mon')
+        def dummy():
+            pass
+
+        self.assertTrue(ScheduledTask.objects.all().exists())
+
+    def test_schedule_task_with_hour(self):
+        """Test decorator registers task with hour argument."""
+        @schedule_task('mon', hour=12)
+        def dummy():
+            pass
+
+        self.assertTrue(ScheduledTask.objects.all().exists())