Add periodic exchange task management API

src/apps/exchange/serializers.py

@@ -1,9 +1,53 @@
 """Сериализаторы приложения обмена данными."""
 
+import json
+from typing import Any
+
 from apps.exchange.models import ExchangeConnection
+from django_celery_beat.models import IntervalSchedule, PeriodicTask
 from rest_framework import serializers
 
 
+def validate_exchange_copy_payload(attrs: dict[str, Any]) -> dict[str, Any]:
+    """Проверить совместимость параметров запуска копирования."""
+    mode = attrs["mode"]
+    table = attrs.get("table")
+    tables = attrs.get("tables")
+
+    if mode == "single" and not table:
+        raise serializers.ValidationError(
+            {"table": "Для mode=single нужно указать table"}
+        )
+
+    if mode == "selected" and not tables:
+        raise serializers.ValidationError(
+            {"tables": "Для mode=selected нужно указать tables"}
+        )
+
+    if mode != "single" and table:
+        raise serializers.ValidationError(
+            {"table": "Поле table допустимо только для mode=single"}
+        )
+
+    if mode != "selected" and tables:
+        raise serializers.ValidationError(
+            {"tables": "Поле tables допустимо только для mode=selected"}
+        )
+
+    return attrs
+
+
+def get_periodic_task_payload(task: PeriodicTask) -> dict[str, Any]:
+    """Извлечь payload exchange-задачи из kwargs django_celery_beat."""
+    try:
+        kwargs = json.loads(task.kwargs or "{}")
+    except json.JSONDecodeError:
+        return {}
+
+    payload = kwargs.get("payload")
+    return payload if isinstance(payload, dict) else {}
+
+
 class ExchangeConnectionSerializer(serializers.ModelSerializer):
     """Сериализатор подключения без выдачи пароля в ответах."""
 
@@ -61,28 +105,257 @@ class ExchangeCopyRequestSerializer(serializers.Serializer):
     truncate_before_copy = serializers.BooleanField(default=True)
 
     def validate(self, attrs):
-        mode = attrs["mode"]
-        table = attrs.get("table")
-        tables = attrs.get("tables")
+        return validate_exchange_copy_payload(attrs)
 
-        if mode == "single" and not table:
+
+class ExchangePeriodicTaskSerializer(serializers.ModelSerializer):
+    """Сериализатор периодической задачи обмена."""
+
+    schedule_type = serializers.SerializerMethodField()
+    interval_every = serializers.SerializerMethodField()
+    interval_period = serializers.SerializerMethodField()
+    crontab_minute = serializers.SerializerMethodField()
+    crontab_hour = serializers.SerializerMethodField()
+    crontab_day_of_week = serializers.SerializerMethodField()
+    crontab_day_of_month = serializers.SerializerMethodField()
+    crontab_month_of_year = serializers.SerializerMethodField()
+    crontab_timezone = serializers.SerializerMethodField()
+    mode = serializers.SerializerMethodField()
+    table = serializers.SerializerMethodField()
+    tables = serializers.SerializerMethodField()
+    truncate_before_copy = serializers.SerializerMethodField()
+
+    class Meta:
+        model = PeriodicTask
+        fields = [
+            "id",
+            "name",
+            "description",
+            "enabled",
+            "schedule_type",
+            "interval_every",
+            "interval_period",
+            "crontab_minute",
+            "crontab_hour",
+            "crontab_day_of_week",
+            "crontab_day_of_month",
+            "crontab_month_of_year",
+            "crontab_timezone",
+            "mode",
+            "table",
+            "tables",
+            "truncate_before_copy",
+            "last_run_at",
+            "total_run_count",
+            "date_changed",
+        ]
+        read_only_fields = fields
+
+    def get_schedule_type(self, obj: PeriodicTask) -> str | None:
+        if obj.interval_id:
+            return "interval"
+        if obj.crontab_id:
+            return "crontab"
+        return None
+
+    def get_interval_every(self, obj: PeriodicTask) -> int | None:
+        return obj.interval.every if obj.interval_id else None
+
+    def get_interval_period(self, obj: PeriodicTask) -> str | None:
+        return obj.interval.period if obj.interval_id else None
+
+    def get_crontab_minute(self, obj: PeriodicTask) -> str | None:
+        return obj.crontab.minute if obj.crontab_id else None
+
+    def get_crontab_hour(self, obj: PeriodicTask) -> str | None:
+        return obj.crontab.hour if obj.crontab_id else None
+
+    def get_crontab_day_of_week(self, obj: PeriodicTask) -> str | None:
+        return obj.crontab.day_of_week if obj.crontab_id else None
+
+    def get_crontab_day_of_month(self, obj: PeriodicTask) -> str | None:
+        return obj.crontab.day_of_month if obj.crontab_id else None
+
+    def get_crontab_month_of_year(self, obj: PeriodicTask) -> str | None:
+        return obj.crontab.month_of_year if obj.crontab_id else None
+
+    def get_crontab_timezone(self, obj: PeriodicTask) -> str | None:
+        if not obj.crontab_id:
+            return None
+        timezone = obj.crontab.timezone
+        return str(timezone) if timezone is not None else None
+
+    def get_mode(self, obj: PeriodicTask) -> str | None:
+        return get_periodic_task_payload(obj).get("mode")
+
+    def get_table(self, obj: PeriodicTask) -> str | None:
+        return get_periodic_task_payload(obj).get("table")
+
+    def get_tables(self, obj: PeriodicTask) -> list[str] | None:
+        return get_periodic_task_payload(obj).get("tables")
+
+    def get_truncate_before_copy(self, obj: PeriodicTask) -> bool | None:
+        return get_periodic_task_payload(obj).get("truncate_before_copy")
+
+
+class ExchangePeriodicTaskUpsertSerializer(serializers.Serializer):
+    """Входные данные для создания и изменения периодической задачи обмена."""
+
+    name = serializers.CharField(max_length=200, required=False)
+    description = serializers.CharField(required=False, allow_blank=True)
+    enabled = serializers.BooleanField(required=False)
+
+    schedule_type = serializers.ChoiceField(
+        choices=["interval", "crontab"],
+        required=False,
+    )
+    interval_every = serializers.IntegerField(min_value=1, required=False)
+    interval_period = serializers.ChoiceField(
+        choices=[choice[0] for choice in IntervalSchedule.PERIOD_CHOICES],
+        required=False,
+    )
+    crontab_minute = serializers.CharField(max_length=64, required=False)
+    crontab_hour = serializers.CharField(max_length=64, required=False)
+    crontab_day_of_week = serializers.CharField(max_length=64, required=False)
+    crontab_day_of_month = serializers.CharField(max_length=64, required=False)
+    crontab_month_of_year = serializers.CharField(max_length=64, required=False)
+
+    mode = serializers.ChoiceField(
+        choices=["all", "single", "selected"],
+        required=False,
+    )
+    table = serializers.CharField(required=False)
+    tables = serializers.ListField(
+        child=serializers.CharField(),
+        required=False,
+        allow_empty=False,
+    )
+    truncate_before_copy = serializers.BooleanField(required=False)
+
+    def validate(self, attrs: dict[str, Any]) -> dict[str, Any]:
+        if not self.instance and "name" not in attrs:
+            raise serializers.ValidationError({"name": "Обязательное поле."})
+
+        schedule_type = self._resolve_schedule_type(attrs)
+        if not schedule_type:
             raise serializers.ValidationError(
-                {"table": "Для mode=single нужно указать table"}
+                {"schedule_type": "Нужно указать тип расписания."}
             )
 
-        if mode == "selected" and not tables:
-            raise serializers.ValidationError(
-                {"tables": "Для mode=selected нужно указать tables"}
-            )
-
-        if mode != "single" and table:
-            raise serializers.ValidationError(
-                {"table": "Поле table допустимо только для mode=single"}
-            )
-
-        if mode != "selected" and tables:
-            raise serializers.ValidationError(
-                {"tables": "Поле tables допустимо только для mode=selected"}
-            )
+        payload = self._build_payload(attrs)
+        schedule = self._build_schedule(attrs, schedule_type)
 
+        attrs["payload"] = validate_exchange_copy_payload(payload)
+        attrs["schedule"] = schedule
         return attrs
+
+    def _resolve_schedule_type(self, attrs: dict[str, Any]) -> str | None:
+        if "schedule_type" in attrs:
+            return attrs["schedule_type"]
+
+        if self.instance and self.instance.interval_id:
+            return "interval"
+
+        if self.instance and self.instance.crontab_id:
+            return "crontab"
+
+        return None
+
+    def _build_payload(self, attrs: dict[str, Any]) -> dict[str, Any]:
+        current_payload = (
+            get_periodic_task_payload(self.instance) if self.instance else {}
+        )
+        mode = attrs.get("mode", current_payload.get("mode", "all"))
+        truncate_before_copy = attrs.get(
+            "truncate_before_copy",
+            current_payload.get("truncate_before_copy", True),
+        )
+
+        if "mode" in attrs and attrs["mode"] != "single" and "table" not in attrs:
+            table = None
+        else:
+            table = attrs.get("table", current_payload.get("table"))
+
+        if "mode" in attrs and attrs["mode"] != "selected" and "tables" not in attrs:
+            tables = None
+        else:
+            tables = attrs.get("tables", current_payload.get("tables"))
+
+        return {
+            "mode": mode,
+            "table": table,
+            "tables": tables,
+            "truncate_before_copy": truncate_before_copy,
+        }
+
+    def _build_schedule(
+        self,
+        attrs: dict[str, Any],
+        schedule_type: str,
+    ) -> dict[str, Any]:
+        if schedule_type == "interval":
+            return self._build_interval_schedule(attrs)
+        return self._build_crontab_schedule(attrs)
+
+    def _build_interval_schedule(self, attrs: dict[str, Any]) -> dict[str, Any]:
+        current_schedule = self.instance.interval if self.instance else None
+        interval_every = attrs.get(
+            "interval_every",
+            current_schedule.every if current_schedule else None,
+        )
+        interval_period = attrs.get(
+            "interval_period",
+            current_schedule.period if current_schedule else None,
+        )
+
+        errors = {}
+        if interval_every is None:
+            errors["interval_every"] = "Обязательное поле для interval."
+        if interval_period is None:
+            errors["interval_period"] = "Обязательное поле для interval."
+        if errors:
+            raise serializers.ValidationError(errors)
+
+        return {
+            "type": "interval",
+            "every": interval_every,
+            "period": interval_period,
+        }
+
+    def _build_crontab_schedule(self, attrs: dict[str, Any]) -> dict[str, Any]:
+        current_schedule = self.instance.crontab if self.instance else None
+        fields = {
+            "minute": attrs.get(
+                "crontab_minute",
+                current_schedule.minute if current_schedule else None,
+            ),
+            "hour": attrs.get(
+                "crontab_hour",
+                current_schedule.hour if current_schedule else None,
+            ),
+            "day_of_week": attrs.get(
+                "crontab_day_of_week",
+                current_schedule.day_of_week if current_schedule else None,
+            ),
+            "day_of_month": attrs.get(
+                "crontab_day_of_month",
+                current_schedule.day_of_month if current_schedule else None,
+            ),
+            "month_of_year": attrs.get(
+                "crontab_month_of_year",
+                current_schedule.month_of_year if current_schedule else None,
+            ),
+        }
+
+        errors = {
+            f"crontab_{field_name}": "Обязательное поле для crontab."
+            for field_name, value in fields.items()
+            if value is None
+        }
+        if errors:
+            raise serializers.ValidationError(errors)
+
+        return {
+            "type": "crontab",
+            **fields,
+        }

src/apps/exchange/services.py

@@ -2,13 +2,17 @@
 
 from __future__ import annotations
 
+import json
 from contextlib import suppress
 from typing import Any
 
 from apps.exchange.models import ExchangeConnection
 from django.apps import apps as django_apps
-from django.db import connections, transaction
+from django.conf import settings
+from django.core.exceptions import ValidationError as DjangoValidationError
+from django.db import IntegrityError, connections, transaction
 from django.utils import timezone
+from django_celery_beat.models import CrontabSchedule, IntervalSchedule, PeriodicTask
 
 
 class ExchangeServiceError(ValueError):
@@ -481,3 +485,163 @@ class ExchangeConnectionService:
             connection.save(
                 update_fields=["last_checked_at", "last_error", "updated_at"]
             )
+
+
+class ExchangePeriodicTaskService:
+    """Сервис управления периодическими задачами обмена."""
+
+    TASK_NAME = "apps.exchange.tasks.dispatch_periodic_exchange_copy"
+
+    @classmethod
+    def get_queryset(cls):
+        return (
+            PeriodicTask.objects.filter(task=cls.TASK_NAME)
+            .select_related("interval", "crontab")
+            .order_by("name")
+        )
+
+    @classmethod
+    @transaction.atomic
+    def create_periodic_task(
+        cls,
+        *,
+        name: str,
+        payload: dict[str, Any],
+        schedule: dict[str, Any],
+        description: str = "",
+        enabled: bool = True,
+    ) -> PeriodicTask:
+        task = PeriodicTask(
+            name=name,
+            task=cls.TASK_NAME,
+            kwargs=json.dumps({"payload": payload}, ensure_ascii=False),
+            description=description,
+            enabled=enabled,
+        )
+        cls._assign_schedule(task=task, schedule=schedule)
+        return cls._save_task(task)
+
+    @classmethod
+    @transaction.atomic
+    def update_periodic_task(
+        cls,
+        *,
+        task: PeriodicTask,
+        payload: dict[str, Any],
+        schedule: dict[str, Any],
+        name: str | None = None,
+        description: str | None = None,
+        enabled: bool | None = None,
+    ) -> PeriodicTask:
+        old_interval_id = task.interval_id
+        old_crontab_id = task.crontab_id
+
+        if name is not None:
+            task.name = name
+        if description is not None:
+            task.description = description
+        if enabled is not None:
+            task.enabled = enabled
+
+        task.kwargs = json.dumps({"payload": payload}, ensure_ascii=False)
+        cls._assign_schedule(task=task, schedule=schedule)
+        task = cls._save_task(task)
+
+        cls._cleanup_unused_interval(old_interval_id)
+        cls._cleanup_unused_crontab(old_crontab_id)
+        return task
+
+    @classmethod
+    def _assign_schedule(cls, *, task: PeriodicTask, schedule: dict[str, Any]) -> None:
+        if schedule["type"] == "interval":
+            task.interval = cls._get_or_create_interval(schedule)
+            task.crontab = None
+            return
+
+        task.crontab = cls._get_or_create_crontab(schedule)
+        task.interval = None
+
+    @classmethod
+    def _get_or_create_interval(cls, schedule: dict[str, Any]) -> IntervalSchedule:
+        interval = IntervalSchedule(
+            every=schedule["every"],
+            period=schedule["period"],
+        )
+        cls._validate_model(interval)
+        interval, _ = IntervalSchedule.objects.get_or_create(
+            every=interval.every,
+            period=interval.period,
+        )
+        return interval
+
+    @classmethod
+    def _get_or_create_crontab(cls, schedule: dict[str, Any]) -> CrontabSchedule:
+        crontab = CrontabSchedule(
+            minute=schedule["minute"],
+            hour=schedule["hour"],
+            day_of_week=schedule["day_of_week"],
+            day_of_month=schedule["day_of_month"],
+            month_of_year=schedule["month_of_year"],
+            timezone=settings.TIME_ZONE,
+        )
+        cls._validate_model(crontab)
+        crontab, _ = CrontabSchedule.objects.get_or_create(
+            minute=crontab.minute,
+            hour=crontab.hour,
+            day_of_week=crontab.day_of_week,
+            day_of_month=crontab.day_of_month,
+            month_of_year=crontab.month_of_year,
+            timezone=crontab.timezone,
+        )
+        return crontab
+
+    @classmethod
+    def _save_task(cls, task: PeriodicTask) -> PeriodicTask:
+        try:
+            task.full_clean()
+            task.save()
+        except DjangoValidationError as exc:
+            raise ExchangeServiceError(cls._format_validation_error(exc)) from exc
+        except IntegrityError as exc:
+            raise ExchangeServiceError(
+                "Периодическая задача с таким именем уже существует"
+            ) from exc
+
+        return task
+
+    @classmethod
+    def _validate_model(cls, instance) -> None:
+        try:
+            instance.full_clean()
+        except DjangoValidationError as exc:
+            raise ExchangeServiceError(cls._format_validation_error(exc)) from exc
+
+    @classmethod
+    def _format_validation_error(cls, exc: DjangoValidationError) -> str:
+        if hasattr(exc, "message_dict"):
+            messages = []
+            for field, field_errors in exc.message_dict.items():
+                messages.extend(f"{field}: {error}" for error in field_errors)
+            return "; ".join(messages)
+
+        return "; ".join(exc.messages)
+
+    @classmethod
+    def _cleanup_unused_interval(cls, interval_id: int | None) -> None:
+        if not interval_id:
+            return
+
+        if PeriodicTask.objects.filter(interval_id=interval_id).exists():
+            return
+
+        IntervalSchedule.objects.filter(id=interval_id).delete()
+
+    @classmethod
+    def _cleanup_unused_crontab(cls, crontab_id: int | None) -> None:
+        if not crontab_id:
+            return
+
+        if PeriodicTask.objects.filter(crontab_id=crontab_id).exists():
+            return
+
+        CrontabSchedule.objects.filter(id=crontab_id).delete()

src/apps/exchange/tasks.py

@@ -7,6 +7,7 @@ import uuid
 from typing import Any
 
 from apps.core.services import BackgroundJobService
+from apps.core.tasks import PeriodicTask as CorePeriodicTask
 from apps.exchange.models import ExchangeConnection
 from apps.exchange.services import ExchangeConnectionService
 from celery import shared_task
@@ -14,6 +15,26 @@ from celery import shared_task
 logger = logging.getLogger(__name__)
 
 
+@shared_task(bind=True, base=CorePeriodicTask)
+def dispatch_periodic_exchange_copy(
+    self,
+    *,
+    payload: dict[str, Any],
+) -> dict[str, Any]:
+    """Поставить в очередь периодическое копирование через активное подключение."""
+    active_connection = ExchangeConnectionService.get_active_connection()
+    task = copy_parsers_data_async.delay(
+        connection_id=active_connection.id,
+        payload=payload,
+        requested_by_id=None,
+    )
+    return {
+        "status": "queued",
+        "task_id": task.id,
+        "connection_id": active_connection.id,
+    }
+
+
 @shared_task(bind=True)
 def copy_parsers_data_async(
     self,

src/apps/exchange/urls.py

@@ -4,6 +4,8 @@ from apps.exchange.views import (
     ExchangeConnectionListCreateView,
     ExchangeConnectionTestView,
     ExchangeCopyDataView,
+    ExchangePeriodicTaskDetailView,
+    ExchangePeriodicTaskListCreateView,
 )
 from django.urls import path
 
@@ -19,6 +21,16 @@ exchange_urlpatterns = [
         name="connections-test",
     ),
     path("copy/", ExchangeCopyDataView.as_view(), name="copy"),
+    path(
+        "periodic-tasks/",
+        ExchangePeriodicTaskListCreateView.as_view(),
+        name="periodic-tasks",
+    ),
+    path(
+        "periodic-tasks/<int:task_id>/",
+        ExchangePeriodicTaskDetailView.as_view(),
+        name="periodic-task-detail",
+    ),
 ]
 
 urlpatterns = []

src/apps/exchange/views.py

@@ -10,10 +10,17 @@ from apps.exchange.serializers import (
     ExchangeConnectionCreateSerializer,
     ExchangeConnectionSerializer,
     ExchangeCopyRequestSerializer,
+    ExchangePeriodicTaskSerializer,
+    ExchangePeriodicTaskUpsertSerializer,
+)
+from apps.exchange.services import (
+    ExchangeConnectionService,
+    ExchangePeriodicTaskService,
+    ExchangeServiceError,
 )
-from apps.exchange.services import ExchangeConnectionService, ExchangeServiceError
 from apps.exchange.tasks import copy_parsers_data_async
 from django.db import IntegrityError
+from django.shortcuts import get_object_or_404
 from drf_yasg import openapi
 from drf_yasg.utils import swagger_auto_schema
 from rest_framework import status
@@ -203,3 +210,124 @@ class ExchangeCopyDataView(APIView):
             },
             status_code=status.HTTP_202_ACCEPTED,
         )
+
+
+class ExchangePeriodicTaskListCreateView(APIView):
+    """API списка и создания периодических задач обмена."""
+
+    permission_classes = [IsAdminUser]
+
+    @swagger_auto_schema(
+        tags=[EXCHANGE_TAG],
+        operation_summary="Список периодических задач обмена",
+        operation_description=(
+            "Возвращает периодические задачи exchange, созданные через "
+            "django_celery_beat."
+        ),
+        responses={
+            200: ExchangePeriodicTaskSerializer(many=True),
+            **ErrorResponses.ADMIN,
+        },
+    )
+    def get(self, request):
+        queryset = ExchangePeriodicTaskService.get_queryset()
+        serializer = ExchangePeriodicTaskSerializer(queryset, many=True)
+        return api_response(serializer.data, status_code=status.HTTP_200_OK)
+
+    @swagger_auto_schema(
+        tags=[EXCHANGE_TAG],
+        operation_summary="Создать периодическую задачу обмена",
+        operation_description=(
+            "Создаёт периодическую задачу exchange с interval или crontab "
+            "расписанием. При выполнении задача использует текущее активное "
+            "подключение exchange."
+        ),
+        request_body=ExchangePeriodicTaskUpsertSerializer,
+        responses={
+            201: ExchangePeriodicTaskSerializer,
+            400: CommonResponses.BAD_REQUEST,
+            **ErrorResponses.ADMIN,
+        },
+    )
+    def post(self, request):
+        serializer = ExchangePeriodicTaskUpsertSerializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+
+        try:
+            task = ExchangePeriodicTaskService.create_periodic_task(
+                name=serializer.validated_data["name"],
+                description=serializer.validated_data.get("description", ""),
+                enabled=serializer.validated_data.get("enabled", True),
+                payload=serializer.validated_data["payload"],
+                schedule=serializer.validated_data["schedule"],
+            )
+        except ExchangeServiceError as exc:
+            raise ValidationError({"periodic_task": str(exc)}) from exc
+
+        output = ExchangePeriodicTaskSerializer(task)
+        return api_created_response(output.data)
+
+
+class ExchangePeriodicTaskDetailView(APIView):
+    """API чтения и изменения периодической задачи обмена."""
+
+    permission_classes = [IsAdminUser]
+
+    @swagger_auto_schema(
+        tags=[EXCHANGE_TAG],
+        operation_summary="Детали периодической задачи обмена",
+        responses={
+            200: ExchangePeriodicTaskSerializer,
+            404: CommonResponses.NOT_FOUND,
+            **ErrorResponses.ADMIN,
+        },
+    )
+    def get(self, request, task_id: int):
+        task = get_object_or_404(
+            ExchangePeriodicTaskService.get_queryset(),
+            id=task_id,
+        )
+        output = ExchangePeriodicTaskSerializer(task)
+        return api_response(output.data, status_code=status.HTTP_200_OK)
+
+    @swagger_auto_schema(
+        tags=[EXCHANGE_TAG],
+        operation_summary="Изменить периодическую задачу обмена",
+        operation_description=(
+            "Обновляет расписание, payload и состояние exchange-задачи. "
+            "PATCH допускает частичное обновление."
+        ),
+        request_body=ExchangePeriodicTaskUpsertSerializer,
+        responses={
+            200: ExchangePeriodicTaskSerializer,
+            400: CommonResponses.BAD_REQUEST,
+            404: CommonResponses.NOT_FOUND,
+            **ErrorResponses.ADMIN,
+        },
+    )
+    def patch(self, request, task_id: int):
+        task = get_object_or_404(
+            ExchangePeriodicTaskService.get_queryset(),
+            id=task_id,
+        )
+        serializer = ExchangePeriodicTaskUpsertSerializer(
+            task,
+            data=request.data,
+            partial=True,
+        )
+        serializer.is_valid(raise_exception=True)
+
+        try:
+            task = ExchangePeriodicTaskService.update_periodic_task(
+                task=task,
+                name=serializer.validated_data.get("name"),
+                description=serializer.validated_data.get("description"),
+                enabled=serializer.validated_data.get("enabled"),
+                payload=serializer.validated_data["payload"],
+                schedule=serializer.validated_data["schedule"],
+            )
+        except ExchangeServiceError as exc:
+            raise ValidationError({"periodic_task": str(exc)}) from exc
+
+        output = ExchangePeriodicTaskSerializer(task)
+        return api_response(output.data, status_code=status.HTTP_200_OK)