create update schedule through view

tubearchivist/task/src/config_schedule.py

@@ -6,6 +6,7 @@ Functionality:
 from datetime import datetime
 
 from appsettings.src.config import AppConfig
+from celery.schedules import crontab
 from common.src.env_settings import EnvironmentSettings
 from django.utils import dateformat
 from django_celery_beat.models import CrontabSchedule
@@ -24,33 +25,24 @@ class ScheduleBuilder:
         "run_backup": "0 18 0",
         "version_check": "0 11 *",
     }
-    CONFIG = {
-        "check_reindex_days": "check_reindex",
-        "run_backup_rotate": "run_backup",
-        "update_subscribed_notify": "update_subscribed",
-        "download_pending_notify": "download_pending",
-        "check_reindex_notify": "check_reindex",
-    }
     MSG = "message:setting"
 
     def __init__(self):
         self.config = AppConfig().config
 
-    def update_schedule_conf(self, form_post):
+    def update_schedule(
-        """process form post, schedules need to be validated before"""
+        self, task_name: str, cron_schedule: str, schedule_conf: dict | None
-        for key, value in form_post.items():
+    ) -> None:
-            if not value:
+        """update schedule"""
-                continue
+        if cron_schedule == "auto":
+            cron_schedule = self.SCHEDULES[task_name]
 
-            if key in self.SCHEDULES:
+        if cron_schedule:
-                if value == "auto":
+            _ = self.get_set_task(task_name, cron_schedule)
-                    value = self.SCHEDULES.get(key)
 
-                _ = self.get_set_task(key, value)
+        if schedule_conf:
-                continue
+            for key, value in schedule_conf.items():
-
+                self.set_config(task_name, key, value)
-            if key in self.CONFIG:
-                self.set_config(key, value)
 
     def get_set_task(self, task_name, schedule=False):
         """get task"""
@@ -73,21 +65,77 @@ class ScheduleBuilder:
         return task
 
     @staticmethod
-    def get_set_cron_tab(schedule):
+    def get_set_cron_tab(schedule: str) -> CrontabSchedule:
         """needs to be validated before"""
         kwargs = dict(zip(["minute", "hour", "day_of_week"], schedule.split()))
         kwargs.update({"timezone": EnvironmentSettings.TZ})
-        crontab, _ = CrontabSchedule.objects.get_or_create(**kwargs)
+        task_crontab, _ = CrontabSchedule.objects.get_or_create(**kwargs)
 
-        return crontab
+        return task_crontab
 
-    def set_config(self, key, value):
+    def set_config(self, task_name: str, key: str, value) -> None:
-        """set task_config"""
+        """set task_config, validate before"""
-        task_name = self.CONFIG.get(key)
+        try:
-        if not task_name:
+            task = CustomPeriodicTask.objects.get(name=task_name)
-            raise ValueError("invalid config key")
+            task.task_config.update({key: value})
+            task.save()
+        except CustomPeriodicTask.DoesNotExist:
+            pass
 
-        task = CustomPeriodicTask.objects.get(name=task_name)
+
-        config_key = key.split(f"{task_name}_")[-1]
+class CrontabValidator:
-        task.task_config.update({config_key: value})
+    """validate crontab"""
-        task.save()
+
+    CONFIG = {
+        "check_reindex": ["days"],
+        "run_backup": ["rotate"],
+    }
+
+    @staticmethod
+    def validate_fields(cron_fields: str) -> None:
+        """expect 3 cron fields"""
+        if not len(cron_fields) == 3:
+            raise ValueError("expected three cron schedule fields")
+
+    @staticmethod
+    def validate_minute(minute_field: str):
+        """expect minute int"""
+        if not minute_field.isdigit():
+            raise ValueError("Invalid value for minutes. Must be an integer.")
+
+        minutes = int(minute_field)
+        if not 0 <= minutes <= 59:
+            raise ValueError("Invalid minutes. Must be between 0 and 59.")
+
+    @staticmethod
+    def validate_cron_tab(minute, hour, day_of_week):
+        """check if crontab can be created"""
+        try:
+            crontab(minute=minute, hour=hour, day_of_week=day_of_week)
+        except ValueError as err:
+            raise ValueError(f"invalid crontab: {err}") from err
+
+    def validate_cron(self, cron_expression):
+        """create crontab schedule"""
+        if not cron_expression or cron_expression == "auto":
+            return
+
+        cron_fields = cron_expression.split()
+        self.validate_fields(cron_fields)
+
+        minute, hour, day_of_week = cron_fields
+        self.validate_minute(minute)
+        self.validate_cron_tab(minute, hour, day_of_week)
+
+    def validate_config(self, task_name: str, schedule_config: dict):
+        """validate config for given task"""
+        if not schedule_config:
+            return
+
+        config_keys = self.CONFIG.get(task_name)
+        if not config_keys:
+            raise ValueError(f"task '{task_name}' doesn't take config")
+
+        for key in schedule_config:
+            if key not in config_keys:
+                raise ValueError(f"invalid config key for task '{task_name}'")

tubearchivist/task/urls.py

@@ -20,7 +20,7 @@ urlpatterns = [
         name="api-task-id",
     ),
     path(
-        "schedule/",
+        "schedule/<slug:task_name>/",
         views.ScheduleView.as_view(),
         name="api-schedule",
     ),

tubearchivist/task/views.py

@@ -1,8 +1,10 @@
 """all task API views"""
 
 from common.views_base import AdminOnly, ApiBaseView
+from django.shortcuts import get_object_or_404
 from rest_framework.response import Response
 from task.models import CustomPeriodicTask
+from task.src.config_schedule import CrontabValidator, ScheduleBuilder
 from task.src.notify import Notifications, get_all_notifications
 from task.src.task_config import TASK_CONFIG
 from task.src.task_manager import TaskCommand, TaskManager
@@ -118,20 +120,40 @@ class TaskIDView(ApiBaseView):
 
 class ScheduleView(ApiBaseView):
     """resolves to /api/task/schedule/<task-name>/
+    POST: create/update schedule for task with config
+    - example: {"schedule": "0 0 *", "config": {"days": 90}}
     DEL: delete schedule for task
     """
 
     permission_classes = [AdminOnly]
 
-    def delete(self, request):
+    def post(self, request, task_name):
-        """delete schedule by task_name query"""
+        """create/update schedule for task"""
-        task_name = request.data.get("task_name")
+        cron_schedule = request.data.get("schedule")
-        try:
+        schedule_config = request.data.get("config")
-            task = CustomPeriodicTask.objects.get(name=task_name)
+        if not cron_schedule and not schedule_config:
-        except CustomPeriodicTask.DoesNotExist:
+            message = {"message": "expected schedule or config key"}
-            message = {"message": "task_name not found"}
+            return Response(message, status=400)
-            return Response(message, status=404)
 
+        try:
+            validator = CrontabValidator()
+            validator.validate_cron(cron_schedule)
+            validator.validate_config(task_name, schedule_config)
+        except ValueError as err:
+            return Response({"message": str(err)}, status=400)
+
+        ScheduleBuilder().update_schedule(
+            task_name, cron_schedule, schedule_config
+        )
+        message = f"update schedule for task {task_name}"
+        if schedule_config:
+            message += f" with config {schedule_config}"
+
+        return Response({"message": message})
+
+    def delete(self, request, task_name):
+        """delete schedule by task_name query"""
+        task = get_object_or_404(CustomPeriodicTask, name=task_name)
         _ = task.delete()
 
         return Response({"success": True})