diff --git a/backend/task/models.py b/backend/task/models.py index 81c74bbd..0bcd699b 100644 --- a/backend/task/models.py +++ b/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}" diff --git a/backend/task/serializers.py b/backend/task/serializers.py new file mode 100644 index 00000000..6c16f633 --- /dev/null +++ b/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", + ] diff --git a/backend/task/urls.py b/backend/task/urls.py index e0c7f2f4..66f985c6 100644 --- a/backend/task/urls.py +++ b/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//", views.ScheduleView.as_view(), diff --git a/backend/task/views.py b/backend/task/views.py index 61d0a013..7965fb38 100644 --- a/backend/task/views.py +++ b/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// 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")