public/tubearchivist
1
0
Fork 0
mirror of https://git.vectorsigma.ru/public/tubearchivist.git synced 2026-08-04 20:19:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

serialize tasks

Browse Source
This commit is contained in:
Simon
2025-02-12 11:16:07 +07:00
parent b2b25f24ed
commit 5498958b36
4 changed files with 272 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
backend/common/serializers.py
View File

@@ -52,8 +52,9 @@ class PaginationSerializer(serializers.Serializer):
class AsyncTaskResponseSerializer(serializers.Serializer):
"""serialize new async task"""
message = serializers.CharField()
message = serializers.CharField(required=False)
task_id = serializers.CharField()
status = serializers.CharField(required=False)
filename = serializers.CharField(required=False)

70
backend/task/serializers.py
View File

@@ -1,7 +1,10 @@
"""serializer for tasks"""
# pylint: disable=abstract-method
from rest_framework import serializers
from task.models import CustomPeriodicTask
from task.src.task_config import TASK_CONFIG
class CustomPeriodicTaskSerializer(serializers.ModelSerializer):
@@ -21,3 +24,70 @@ class CustomPeriodicTaskSerializer(serializers.ModelSerializer):
"last_run_at",
"config",
]
class TaskResultSerializer(serializers.Serializer):
"""serialize task result stored in redis"""
status = serializers.ChoiceField(
choices=[
"PENDING",
"STARTED",
"SUCCESS",
"FAILURE",
"RETRY",
"REVOKED",
]
)
result = serializers.CharField(allow_null=True)
traceback = serializers.CharField(allow_null=True)
date_done = serializers.CharField()
name = serializers.CharField()
args = serializers.ListField(child=serializers.JSONField(), required=False)
children = serializers.ListField(
child=serializers.CharField(), required=False
)
kwargs = serializers.DictField(required=False)
worker = serializers.CharField(required=False)
retries = serializers.IntegerField(required=False)
queue = serializers.CharField(required=False)
task_id = serializers.CharField()
class TaskIDDataSerializer(serializers.Serializer):
"""serialize task by ID POST data"""
command = serializers.ChoiceField(choices=["stop", "kill"])
class TaskCreateDataSerializer(serializers.Serializer):
"""serialize task create data"""
schedule = serializers.CharField(required=False)
config = serializers.DictField(required=False)
class TaskNotificationItemSerializer(serializers.Serializer):
"""serialize single task notification"""
urls = serializers.ListField(child=serializers.CharField())
title = serializers.CharField()
def create_dynamic_notification_serializer():
"""use task config"""
fields = {
key: TaskNotificationItemSerializer(required=False)
for key in TASK_CONFIG
}
return type("DynamicDictSerializer", (serializers.Serializer,), fields)
TaskNotificationSerializer = create_dynamic_notification_serializer()
class TaskNotificationPostSerializer(serializers.Serializer):
"""serialize task notification POST"""
task_name = serializers.ChoiceField(choices=list(TASK_CONFIG))
url = serializers.CharField(required=False)

22
backend/task/src/config_schedule.py
View File

@@ -28,18 +28,19 @@ class ScheduleBuilder:
def update_schedule(
self, task_name: str, cron_schedule: str, schedule_conf: dict | None
) -> None:
) -> CustomPeriodicTask:
"""update schedule"""
if cron_schedule == "auto":
cron_schedule = self.SCHEDULES[task_name]
if cron_schedule:
_ = self.get_set_task(task_name, cron_schedule)
task = self.get_set_task(task_name, cron_schedule)
if schedule_conf:
for key, value in schedule_conf.items():
self.set_config(task_name, key, value)
return task
def get_set_task(self, task_name, schedule=False):
"""get task"""
try:
@@ -69,14 +70,15 @@ class ScheduleBuilder:
return task_crontab
def set_config(self, task_name: str, key: str, value) -> None:
def set_config(
self, task_name: str, key: str, value
) -> CustomPeriodicTask:
"""set task_config, validate before"""
try:
task = CustomPeriodicTask.objects.get(name=task_name)
task.task_config.update({key: value})
task.save()
except CustomPeriodicTask.DoesNotExist:
pass
task = CustomPeriodicTask.objects.get(name=task_name)
task.task_config.update({key: value})
task.save()
return task
class CrontabValidator:

254
backend/task/views.py
View File

@@ -1,10 +1,22 @@
"""all task API views"""
from common.serializers import (
AsyncTaskResponseSerializer,
ErrorResponseSerializer,
)
from common.views_base import AdminOnly, ApiBaseView
from django.shortcuts import get_object_or_404
from drf_spectacular.utils import OpenApiResponse, extend_schema
from rest_framework.response import Response
from task.models import CustomPeriodicTask
from task.serializers import CustomPeriodicTaskSerializer
from task.serializers import (
CustomPeriodicTaskSerializer,
TaskCreateDataSerializer,
TaskIDDataSerializer,
TaskNotificationPostSerializer,
TaskNotificationSerializer,
TaskResultSerializer,
)
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
@@ -18,12 +30,14 @@ class TaskListView(ApiBaseView):
permission_classes = [AdminOnly]
@extend_schema(responses=TaskResultSerializer(many=True))
def get(self, request):
"""handle get request"""
"""get all stored task results"""
# pylint: disable=unused-argument
all_results = TaskManager().get_all_results()
serializer = TaskResultSerializer(all_results, many=True)
return Response(all_results)
return Response(serializer.data)
class TaskNameListView(ApiBaseView):
@@ -34,36 +48,55 @@ class TaskNameListView(ApiBaseView):
permission_classes = [AdminOnly]
@extend_schema(
responses={
200: OpenApiResponse(TaskResultSerializer(many=True)),
404: OpenApiResponse(
ErrorResponseSerializer(), description="task name not found"
),
},
)
def get(self, request, task_name):
"""handle get request"""
"""get stored task by name"""
# pylint: disable=unused-argument
if task_name not in TASK_CONFIG:
message = {"message": "invalid task name"}
return Response(message, status=404)
error = ErrorResponseSerializer({"error": "task name not found"})
return Response(error.data, status=404)
all_results = TaskManager().get_tasks_by_name(task_name)
serializer = TaskResultSerializer(all_results, many=True)
return Response(all_results)
return Response(serializer.data)
@extend_schema(
responses={
200: OpenApiResponse(AsyncTaskResponseSerializer()),
400: OpenApiResponse(
ErrorResponseSerializer(), description="bad request"
),
404: OpenApiResponse(
ErrorResponseSerializer(), description="task name not found"
),
}
)
def post(self, request, task_name):
"""
handle post request
404 for invalid task_name
400 if task can't be started here without argument
"""
"""start new task without args"""
# pylint: disable=unused-argument
task_config = TASK_CONFIG.get(task_name)
if not task_config:
message = {"message": "invalid task name"}
return Response(message, status=404)
error = ErrorResponseSerializer({"error": "task name not found"})
return Response(error.data, status=404)
if not task_config.get("api_start"):
message = {"message": "can not start task through this endpoint"}
return Response(message, status=400)
error = ErrorResponseSerializer(
{"error": "can not start task through this endpoint"}
)
return Response(error.data, status=404)
message = TaskCommand().start(task_name)
serializer = AsyncTaskResponseSerializer(message)
return Response({"message": message})
return Response(serializer.data)
class TaskIDView(ApiBaseView):
@@ -75,43 +108,70 @@ class TaskIDView(ApiBaseView):
valid_commands = ["stop", "kill"]
permission_classes = [AdminOnly]
@extend_schema(
responses={
200: OpenApiResponse(TaskResultSerializer()),
404: OpenApiResponse(
ErrorResponseSerializer(), description="task not found"
),
},
)
def get(self, request, task_id):
"""handle get request"""
"""get task by ID"""
# pylint: disable=unused-argument
task_result = TaskManager().get_task(task_id)
if not task_result:
message = {"message": "task id not found"}
return Response(message, status=404)
error = ErrorResponseSerializer({"error": "task ID not found"})
return Response(error.data, status=404)
return Response(task_result)
serializer = TaskResultSerializer(task_result)
return Response(serializer.data)
@extend_schema(
request=TaskIDDataSerializer(),
responses={
204: OpenApiResponse(description="task command sent"),
400: OpenApiResponse(
ErrorResponseSerializer(), description="bad request"
),
404: OpenApiResponse(
ErrorResponseSerializer(), description="task not found"
),
},
)
def post(self, request, task_id):
"""post command to task"""
command = request.data.get("command")
if not command or command not in self.valid_commands:
message = {"message": "no valid command found"}
return Response(message, status=400)
data_serializer = TaskIDDataSerializer(data=request.data)
data_serializer.is_valid(raise_exception=True)
validated_data = data_serializer.validated_data
command = validated_data["command"]
task_result = TaskManager().get_task(task_id)
if not task_result:
message = {"message": "task id not found"}
return Response(message, status=404)
error = ErrorResponseSerializer({"error": "task ID not found"})
return Response(error.data, status=404)
task_conf = TASK_CONFIG.get(task_result.get("name"))
if command == "stop":
if not task_conf.get("api_stop"):
message = {"message": "task can not be stopped"}
return Response(message, status=400)
error = ErrorResponseSerializer(
{"error": "task can not be stopped"}
)
return Response(error.data, status=400)
TaskCommand().stop(task_id)
if command == "kill":
if not task_conf.get("api_stop"):
message = {"message": "task can not be killed"}
return Response(message, status=400)
error = ErrorResponseSerializer(
{"error": "task can not be killed"}
)
return Response(error.data, status=400)
TaskCommand().kill(task_id)
return Response({"message": "command sent"})
return Response(status=204)
class ScheduleListView(ApiBaseView):
@@ -121,11 +181,16 @@ class ScheduleListView(ApiBaseView):
permission_classes = [AdminOnly]
@extend_schema(
responses={
200: OpenApiResponse(CustomPeriodicTaskSerializer(many=True)),
},
)
def get(self, request):
"""get all schedules"""
tasks = CustomPeriodicTask.objects.all()
response = CustomPeriodicTaskSerializer(tasks, many=True).data
return Response(response)
serializer = CustomPeriodicTaskSerializer(tasks, many=True)
return Response(serializer.data)
class ScheduleView(ApiBaseView):
@@ -137,42 +202,77 @@ class ScheduleView(ApiBaseView):
permission_classes = [AdminOnly]
@extend_schema(
responses={
200: OpenApiResponse(CustomPeriodicTaskSerializer()),
404: OpenApiResponse(
ErrorResponseSerializer(), description="schedule not found"
),
},
)
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)
serializer = CustomPeriodicTaskSerializer(task)
return Response(serializer.data)
@extend_schema(
request=TaskCreateDataSerializer(),
responses={
200: OpenApiResponse(CustomPeriodicTaskSerializer()),
400: OpenApiResponse(
ErrorResponseSerializer(), description="bad request"
),
},
)
def post(self, request, task_name):
"""create/update schedule for task"""
cron_schedule = request.data.get("schedule")
schedule_config = request.data.get("config")
data_serializer = TaskCreateDataSerializer(data=request.data)
data_serializer.is_valid(raise_exception=True)
validated_data = data_serializer.validated_data
cron_schedule = validated_data.get("schedule")
schedule_config = validated_data.get("config")
if not cron_schedule and not schedule_config:
message = {"message": "expected schedule or config key"}
return Response(message, status=400)
error = ErrorResponseSerializer(
{"error": "expected schedule or config key"}
)
return Response(error.data, status=400)
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)
error = ErrorResponseSerializer({"error": str(err)})
return Response(error.data, status=400)
ScheduleBuilder().update_schedule(
task = 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})
print(message)
serializer = CustomPeriodicTaskSerializer(task)
return Response(serializer.data)
@extend_schema(
responses={
204: OpenApiResponse(description="schedule deleted"),
404: OpenApiResponse(
ErrorResponseSerializer(), description="schedule not found"
),
},
)
def delete(self, request, task_name):
"""delete schedule by task_name query"""
"""delete schedule by task_name"""
task = get_object_or_404(CustomPeriodicTask, name=task_name)
_ = task.delete()
return Response({"success": True})
return Response(status=204)
class ScheduleNotification(ApiBaseView):
@@ -182,42 +282,64 @@ class ScheduleNotification(ApiBaseView):
DEL: delete notification
"""
@extend_schema(
responses=TaskNotificationSerializer(),
)
def get(self, request):
"""handle get request"""
serializer = TaskNotificationSerializer(get_all_notifications())
return Response(get_all_notifications())
return Response(serializer.data)
@extend_schema(
request=TaskNotificationPostSerializer(),
responses={
200: OpenApiResponse(TaskNotificationSerializer()),
400: OpenApiResponse(
ErrorResponseSerializer(), description="bad request"
),
},
)
def post(self, request):
"""handle create notification"""
task_name = request.data.get("task_name")
url = request.data.get("url")
if not TASK_CONFIG.get(task_name):
message = {"message": "task_name not found"}
return Response(message, status=404)
"""create notification"""
data_serializer = TaskNotificationPostSerializer(data=request.data)
data_serializer.is_valid(raise_exception=True)
validated_data = data_serializer.validated_data
task_name = validated_data["task_name"]
url = validated_data["url"]
if not url:
message = {"message": "missing url key"}
return Response(message, status=400)
error = ErrorResponseSerializer({"error": "missing url"})
return Response(error.data, status=400)
Notifications(task_name).add_url(url)
message = {"task_name": task_name, "url": url}
return Response(message)
serializer = TaskNotificationSerializer(get_all_notifications())
return Response(serializer.data)
@extend_schema(
request=TaskNotificationPostSerializer(),
responses={
204: OpenApiResponse(description="notification url deleted"),
400: OpenApiResponse(
ErrorResponseSerializer(), description="bad request"
),
},
)
def delete(self, request):
"""handle delete"""
"""delete notification"""
task_name = request.data.get("task_name")
url = request.data.get("url")
data_serializer = TaskNotificationPostSerializer(data=request.data)
data_serializer.is_valid(raise_exception=True)
validated_data = data_serializer.validated_data
if not TASK_CONFIG.get(task_name):
message = {"message": "task_name not found"}
return Response(message, status=404)
task_name = validated_data["task_name"]
url = validated_data.get("url")
if url:
response, status_code = Notifications(task_name).remove_url(url)
Notifications(task_name).remove_url(url)
else:
response, status_code = Notifications(task_name).remove_task()
Notifications(task_name).remove_task()
return Response({"response": response, "status_code": status_code})
return Response(status=204)