add get views for schedules, #723

backend/task/models.py

@@ -1,10 +1,19 @@
 """task model"""
 
 from django.db import models
-from django_celery_beat.models import PeriodicTask
+from django_celery_beat.models import PeriodicTask, cronexp
 
 
 class CustomPeriodicTask(PeriodicTask):
     """add custom metadata to task"""
 
     task_config = models.JSONField(default=dict)
+
+    @property
+    def schedule_parsed(self):
+        """parse schedule"""
+        minute = cronexp(self.crontab.minute)
+        hour = cronexp(self.crontab.hour)
+        day_of_week = cronexp(self.crontab.day_of_week)
+
+        return f"{minute} {hour} {day_of_week}"

backend/task/serializers.py

@@ -0,0 +1,23 @@
+"""serializer for tasks"""
+
+from rest_framework import serializers
+from task.models import CustomPeriodicTask
+
+
+class CustomPeriodicTaskSerializer(serializers.ModelSerializer):
+    """serialize CustomPeriodicTask"""
+
+    schedule = serializers.CharField(source="schedule_parsed")
+    schedule_human = serializers.CharField(source="crontab.human_readable")
+    last_run_at = serializers.DateTimeField()
+    config = serializers.DictField(source="task_config")
+
+    class Meta:
+        model = CustomPeriodicTask
+        fields = [
+            "name",
+            "schedule",
+            "schedule_human",
+            "last_run_at",
+            "config",
+        ]

backend/task/urls.py

@@ -19,6 +19,11 @@ urlpatterns = [
         views.TaskIDView.as_view(),
         name="api-task-id",
     ),
+    path(
+        "schedule/",
+        views.ScheduleListView.as_view(),
+        name="api-schedule-list",
+    ),
     path(
         "schedule/<slug:task_name>/",
         views.ScheduleView.as_view(),

backend/task/views.py

@@ -4,6 +4,7 @@ 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.serializers import CustomPeriodicTaskSerializer
 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
@@ -118,6 +119,20 @@ class TaskIDView(ApiBaseView):
         return f"message:{task_conf.get('group')}:{task_id.split('-')[0]}"
 
 
+class ScheduleListView(ApiBaseView):
+    """resolves to /api/task/schedule/
+    GET: list all schedules
+    """
+
+    permission_classes = [AdminOnly]
+
+    def get(self, request):
+        """get all schedules"""
+        tasks = CustomPeriodicTask.objects.all()
+        response = CustomPeriodicTaskSerializer(tasks, many=True).data
+        return Response(response)
+
+
 class ScheduleView(ApiBaseView):
     """resolves to /api/task/schedule/<task-name>/
     POST: create/update schedule for task with config
@@ -127,6 +142,12 @@ class ScheduleView(ApiBaseView):
 
     permission_classes = [AdminOnly]
 
+    def get(self, request, task_name):
+        """get single schedule by task_name"""
+        task = get_object_or_404(CustomPeriodicTask, name=task_name)
+        response = CustomPeriodicTaskSerializer(task).data
+        return Response(response)
+
     def post(self, request, task_name):
         """create/update schedule for task"""
         cron_schedule = request.data.get("schedule")