split task app, migrate scheduler instances

tubearchivist/api/urls.py

@@ -31,31 +31,6 @@ urlpatterns = [
         views.BackupApiView.as_view(),
         name="api-backup",
     ),
-    path(
-        "task-name/",
-        views.TaskListView.as_view(),
-        name="api-task-list",
-    ),
-    path(
-        "task-name/<slug:task_name>/",
-        views.TaskNameListView.as_view(),
-        name="api-task-name-list",
-    ),
-    path(
-        "task-id/<slug:task_id>/",
-        views.TaskIDView.as_view(),
-        name="api-task-id",
-    ),
-    path(
-        "schedule/",
-        views.ScheduleView.as_view(),
-        name="api-schedule",
-    ),
-    path(
-        "schedule/notification/",
-        views.ScheduleNotification.as_view(),
-        name="api-schedule-notification",
-    ),
     path(
         "config/user/",
         views.UserConfigView.as_view(),

tubearchivist/api/views.py

@@ -11,7 +11,6 @@ from api.src.aggs import (
 )
 from api.src.search_processor import SearchProcess
 from download.src.yt_dlp_base import CookieHandler
-from home.models import CustomPeriodicTask
 from home.src.es.backup import ElasticBackup
 from home.src.es.connect import ElasticWrap
 from home.src.es.snapshot import ElasticSnapshot
@@ -20,13 +19,9 @@ from home.src.frontend.watched import WatchState
 from home.src.index.generic import Pagination
 from home.src.index.reindex import ReindexProgress
 from home.src.ta.config import AppConfig, ReleaseVersion
-from home.src.ta.notify import Notifications, get_all_notifications
 from home.src.ta.settings import EnvironmentSettings
 from home.src.ta.ta_redis import RedisArchivist
-from home.src.ta.task_config import TASK_CONFIG
-from home.src.ta.task_manager import TaskCommand, TaskManager
 from home.src.ta.users import UserConfig
-from home.tasks import check_reindex, run_restore_backup
 from rest_framework import permissions
 from rest_framework.authentication import (
     SessionAuthentication,
@@ -36,6 +31,8 @@ from rest_framework.authtoken.models import Token
 from rest_framework.authtoken.views import ObtainAuthToken
 from rest_framework.response import Response
 from rest_framework.views import APIView
+from task.src.task_manager import TaskCommand
+from task.tasks import check_reindex, run_restore_backup
 
 
 def check_admin(user):
@@ -309,163 +306,6 @@ class BackupApiView(ApiBaseView):
         return Response(message)
 
 
-class TaskListView(ApiBaseView):
-    """resolves to /api/task-name/
-    GET: return a list of all stored task results
-    """
-
-    permission_classes = [AdminOnly]
-
-    def get(self, request):
-        """handle get request"""
-        # pylint: disable=unused-argument
-        all_results = TaskManager().get_all_results()
-
-        return Response(all_results)
-
-
-class TaskNameListView(ApiBaseView):
-    """resolves to /api/task-name/<task-name>/
-    GET: return a list of stored results of task
-    POST: start new background process
-    """
-
-    permission_classes = [AdminOnly]
-
-    def get(self, request, task_name):
-        """handle get request"""
-        # pylint: disable=unused-argument
-        if task_name not in TASK_CONFIG:
-            message = {"message": "invalid task name"}
-            return Response(message, status=404)
-
-        all_results = TaskManager().get_tasks_by_name(task_name)
-
-        return Response(all_results)
-
-    def post(self, request, task_name):
-        """
-        handle post request
-        404 for invalid task_name
-        400 if task can't be started here without argument
-        """
-        # 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)
-
-        if not task_config.get("api_start"):
-            message = {"message": "can not start task through this endpoint"}
-            return Response(message, status=400)
-
-        message = TaskCommand().start(task_name)
-
-        return Response({"message": message})
-
-
-class TaskIDView(ApiBaseView):
-    """resolves to /api/task-id/<task-id>/
-    GET: return details of task id
-    """
-
-    valid_commands = ["stop", "kill"]
-    permission_classes = [AdminOnly]
-
-    def get(self, request, task_id):
-        """handle get request"""
-        # 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)
-
-        return Response(task_result)
-
-    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)
-
-        task_result = TaskManager().get_task(task_id)
-        if not task_result:
-            message = {"message": "task id not found"}
-            return Response(message, 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)
-
-            message_key = self._build_message_key(task_conf, task_id)
-            TaskCommand().stop(task_id, message_key)
-        if command == "kill":
-            if not task_conf.get("api_stop"):
-                message = {"message": "task can not be killed"}
-                return Response(message, status=400)
-
-            TaskCommand().kill(task_id)
-
-        return Response({"message": "command sent"})
-
-    def _build_message_key(self, task_conf, task_id):
-        """build message key to forward command to notification"""
-        return f"message:{task_conf.get('group')}:{task_id.split('-')[0]}"
-
-
-class ScheduleView(ApiBaseView):
-    """resolves to /api/schedule/
-    DEL: delete schedule for task
-    """
-
-    permission_classes = [AdminOnly]
-
-    def delete(self, request):
-        """delete schedule by task_name query"""
-        task_name = request.data.get("task_name")
-        try:
-            task = CustomPeriodicTask.objects.get(name=task_name)
-        except CustomPeriodicTask.DoesNotExist:
-            message = {"message": "task_name not found"}
-            return Response(message, status=404)
-
-        _ = task.delete()
-
-        return Response({"success": True})
-
-
-class ScheduleNotification(ApiBaseView):
-    """resolves to /api/schedule/notification/
-    GET: get all schedule notifications
-    DEL: delete notification
-    """
-
-    def get(self, request):
-        """handle get request"""
-
-        return Response(get_all_notifications())
-
-    def delete(self, request):
-        """handle delete"""
-
-        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)
-
-        if url:
-            response, status_code = Notifications(task_name).remove_url(url)
-        else:
-            response, status_code = Notifications(task_name).remove_task()
-
-        return Response({"response": response, "status_code": status_code})
-
-
 class RefreshView(ApiBaseView):
     """resolves to /api/refresh/
     GET: get refresh progress

tubearchivist/channel/views.py

@@ -4,8 +4,8 @@ from api.views import AdminWriteOnly, ApiBaseView
 from channel.src.index import YoutubeChannel
 from download.src.subscriptions import ChannelSubscription
 from home.src.ta.urlparser import Parser
-from home.tasks import subscribe_to
 from rest_framework.response import Response
+from task.tasks import subscribe_to
 
 
 class ChannelApiView(ApiBaseView):

tubearchivist/config/settings.py

@@ -66,6 +66,7 @@ INSTALLED_APPS = [
     "channel",
     "playlist",
     "download",
+    "task",
     "config",
 ]
 

tubearchivist/config/urls.py

@@ -24,5 +24,6 @@ urlpatterns = [
     path("api/channel/", include("channel.urls")),
     path("api/playlist/", include("playlist.urls")),
     path("api/download/", include("download.urls")),
+    path("api/task/", include("task.urls")),
     path("admin/", admin.site.urls),
 ]

tubearchivist/download/views.py

@@ -2,8 +2,8 @@
 
 from api.views import AdminOnly, ApiBaseView
 from download.src.queue import PendingInteract
-from home.tasks import download_pending, extrac_dl
 from rest_framework.response import Response
+from task.tasks import download_pending, extrac_dl
 
 
 class DownloadApiListView(ApiBaseView):

tubearchivist/home/migrations/0003_delete_customperiodictask.py

@@ -0,0 +1,17 @@
+# Generated by Django 5.0.7 on 2024-07-20 10:09
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("home", "0002_customperiodictask"),
+        ("task", "0002_migrate_scheduler"),
+    ]
+
+    operations = [
+        migrations.DeleteModel(
+            name="CustomPeriodicTask",
+        ),
+    ]

tubearchivist/home/models.py

@@ -6,7 +6,6 @@ from django.contrib.auth.models import (
     PermissionsMixin,
 )
 from django.db import models
-from django_celery_beat.models import PeriodicTask
 
 
 class AccountManager(BaseUserManager):
@@ -53,9 +52,3 @@ class Account(AbstractBaseUser, PermissionsMixin):
 
     USERNAME_FIELD = "name"
     REQUIRED_FIELDS = ["password"]
-
-
-class CustomPeriodicTask(PeriodicTask):
-    """add custom metadata to to task"""
-
-    task_config = models.JSONField(default=dict)

tubearchivist/home/src/es/backup.py

@@ -10,11 +10,11 @@ import os
 import zipfile
 from datetime import datetime
 
-from home.models import CustomPeriodicTask
 from home.src.es.connect import ElasticWrap, IndexPaginate
 from home.src.ta.config import AppConfig
 from home.src.ta.helper import get_mapping, ignore_filelist
 from home.src.ta.settings import EnvironmentSettings
+from task.models import CustomPeriodicTask
 
 
 class ElasticBackup:
@@ -139,6 +139,8 @@ class ElasticBackup:
         elif len(file_split) == 3:
             timestamp = file_split[1]
             reason = file_split[2].strip(".zip")
+        else:
+            raise ValueError
 
         data = {
             "filename": filename,

tubearchivist/home/src/frontend/forms_schedule.py

@@ -6,7 +6,7 @@ Functionality:
 
 from celery.schedules import crontab
 from django import forms
-from home.src.ta.task_config import TASK_CONFIG
+from task.src.task_config import TASK_CONFIG
 
 
 class CrontabValidator:

tubearchivist/home/src/index/reindex.py

@@ -14,12 +14,12 @@ from channel.src.index import YoutubeChannel
 from download.src.subscriptions import ChannelSubscription
 from download.src.thumbnails import ThumbManager
 from download.src.yt_dlp_base import CookieHandler
-from home.models import CustomPeriodicTask
 from home.src.es.connect import ElasticWrap, IndexPaginate
 from home.src.ta.config import AppConfig
 from home.src.ta.settings import EnvironmentSettings
 from home.src.ta.ta_redis import RedisQueue
 from playlist.src.index import YoutubePlaylist
+from task.models import CustomPeriodicTask
 from video.src.comments import Comments
 from video.src.index import YoutubeVideo
 

tubearchivist/home/src/ta/notify.py

@@ -2,8 +2,8 @@
 
 import apprise
 from home.src.es.connect import ElasticWrap
-from home.src.ta.task_config import TASK_CONFIG
-from home.src.ta.task_manager import TaskManager
+from task.src.task_config import TASK_CONFIG
+from task.src.task_manager import TaskManager
 
 
 class Notifications:

tubearchivist/home/views.py

@@ -22,7 +22,6 @@ from django.utils.decorators import method_decorator
 from django.views import View
 from download.src.queue import PendingInteract
 from download.src.yt_dlp_base import CookieHandler
-from home.models import CustomPeriodicTask
 from home.src.es.backup import ElasticBackup
 from home.src.es.connect import ElasticWrap
 from home.src.es.snapshot import ElasticSnapshot
@@ -44,15 +43,16 @@ from home.src.frontend.forms_schedule import (
 from home.src.index.generic import Pagination
 from home.src.index.reindex import ReindexProgress
 from home.src.ta.config import AppConfig, ReleaseVersion
-from home.src.ta.config_schedule import ScheduleBuilder
 from home.src.ta.helper import check_stylesheet, time_parser
 from home.src.ta.notify import Notifications, get_all_notifications
 from home.src.ta.settings import EnvironmentSettings
 from home.src.ta.ta_redis import RedisArchivist
 from home.src.ta.users import UserConfig
-from home.tasks import index_channel_playlists, subscribe_to
 from playlist.src.index import YoutubePlaylist
 from rest_framework.authtoken.models import Token
+from task.models import CustomPeriodicTask
+from task.src.config_schedule import ScheduleBuilder
+from task.tasks import index_channel_playlists, subscribe_to
 from video.src.constants import VideoTypeEnum
 
 

tubearchivist/playlist/views.py

@@ -3,10 +3,10 @@
 from api.views import AdminWriteOnly, ApiBaseView
 from download.src.subscriptions import PlaylistSubscription
 from home.src.ta.users import UserConfig
-from home.tasks import subscribe_to
 from playlist.src.index import YoutubePlaylist
 from rest_framework import status
 from rest_framework.response import Response
+from task.tasks import subscribe_to
 
 
 class PlaylistApiListView(ApiBaseView):

tubearchivist/task/migrations/0001_initial.py

@@ -0,0 +1,34 @@
+# Generated by Django 5.0.7 on 2024-07-20 09:54
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        ("django_celery_beat", "0018_improve_crontab_helptext"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="CustomPeriodicTask",
+            fields=[
+                (
+                    "periodictask_ptr",
+                    models.OneToOneField(
+                        auto_created=True,
+                        on_delete=django.db.models.deletion.CASCADE,
+                        parent_link=True,
+                        primary_key=True,
+                        serialize=False,
+                        to="django_celery_beat.periodictask",
+                    ),
+                ),
+                ("task_config", models.JSONField(default=dict)),
+            ],
+            bases=("django_celery_beat.periodictask",),
+        ),
+    ]

tubearchivist/task/migrations/0002_migrate_scheduler.py

@@ -0,0 +1,31 @@
+"""custom migration to copy CustomPeriodicTask to new task app"""
+
+from django.db import migrations
+
+def copy_data(apps, schema_editor):
+    """copy between apps"""
+
+    OldCustomPeriodicTask = apps.get_model("home", "CustomPeriodicTask")
+    NewCustomPeriodicTask = apps.get_model("task", "CustomPeriodicTask")
+
+    for old_instance in OldCustomPeriodicTask.objects.all():
+        field_data = {
+            field.name: getattr(old_instance, field.name)
+            for field in OldCustomPeriodicTask._meta.fields
+        }
+        field_data.pop("id", None)
+        new_instance = NewCustomPeriodicTask(**field_data)
+        new_instance.save()
+
+
+class Migration(migrations.Migration):
+    """migration"""
+
+    dependencies = [
+        ("task", "0001_initial"),
+        ("home", "0002_customperiodictask"),
+    ]
+
+    operations = [
+        migrations.RunPython(copy_data),
+    ]

tubearchivist/task/models.py

@@ -0,0 +1,10 @@
+"""task model"""
+
+from django.db import models
+from django_celery_beat.models import PeriodicTask
+
+
+class CustomPeriodicTask(PeriodicTask):
+    """add custom metadata to task"""
+
+    task_config = models.JSONField(default=dict)

tubearchivist/task/src/config_schedule.py

@@ -7,10 +7,10 @@ from datetime import datetime
 
 from django.utils import dateformat
 from django_celery_beat.models import CrontabSchedule
-from home.models import CustomPeriodicTask
 from home.src.ta.config import AppConfig
 from home.src.ta.settings import EnvironmentSettings
-from home.src.ta.task_config import TASK_CONFIG
+from task.models import CustomPeriodicTask
+from task.src.task_config import TASK_CONFIG
 
 
 class ScheduleBuilder:

tubearchivist/task/src/task_manager.py

@@ -6,7 +6,7 @@ functionality:
 
 from home.celery import app as celery_app
 from home.src.ta.ta_redis import RedisArchivist, TaskRedis
-from home.src.ta.task_config import TASK_CONFIG
+from task.src.task_config import TASK_CONFIG
 
 
 class TaskManager:

tubearchivist/task/tasks.py

@@ -21,9 +21,9 @@ from home.src.index.reindex import Reindex, ReindexManual, ReindexPopulate
 from home.src.ta.config import ReleaseVersion
 from home.src.ta.notify import Notifications
 from home.src.ta.ta_redis import RedisArchivist
-from home.src.ta.task_config import TASK_CONFIG
-from home.src.ta.task_manager import TaskManager
 from home.src.ta.urlparser import Parser
+from task.src.task_config import TASK_CONFIG
+from task.src.task_manager import TaskManager
 
 
 class BaseTask(Task):

tubearchivist/task/urls.py

@@ -0,0 +1,32 @@
+"""all tasks api URLs"""
+
+from django.urls import path
+from task import views
+
+urlpatterns = [
+    path(
+        "by-name/",
+        views.TaskListView.as_view(),
+        name="api-task-list",
+    ),
+    path(
+        "by-name/<slug:task_name>/",
+        views.TaskNameListView.as_view(),
+        name="api-task-name-list",
+    ),
+    path(
+        "by-id/<slug:task_id>/",
+        views.TaskIDView.as_view(),
+        name="api-task-id",
+    ),
+    path(
+        "schedule/",
+        views.ScheduleView.as_view(),
+        name="api-schedule",
+    ),
+    path(
+        "notification/",
+        views.ScheduleNotification.as_view(),
+        name="api-schedule-notification",
+    ),
+]

tubearchivist/task/views.py

@@ -0,0 +1,166 @@
+"""all task API views"""
+
+from api.views import AdminOnly, ApiBaseView
+from home.src.ta.notify import Notifications, get_all_notifications
+from rest_framework.response import Response
+from task.models import CustomPeriodicTask
+from task.src.task_config import TASK_CONFIG
+from task.src.task_manager import TaskCommand, TaskManager
+
+
+class TaskListView(ApiBaseView):
+    """resolves to /api/task/by-name/
+    GET: return a list of all stored task results
+    """
+
+    permission_classes = [AdminOnly]
+
+    def get(self, request):
+        """handle get request"""
+        # pylint: disable=unused-argument
+        all_results = TaskManager().get_all_results()
+
+        return Response(all_results)
+
+
+class TaskNameListView(ApiBaseView):
+    """resolves to /api/task/by-name/<task-name>/
+    GET: return a list of stored results of task
+    POST: start new background process
+    """
+
+    permission_classes = [AdminOnly]
+
+    def get(self, request, task_name):
+        """handle get request"""
+        # pylint: disable=unused-argument
+        if task_name not in TASK_CONFIG:
+            message = {"message": "invalid task name"}
+            return Response(message, status=404)
+
+        all_results = TaskManager().get_tasks_by_name(task_name)
+
+        return Response(all_results)
+
+    def post(self, request, task_name):
+        """
+        handle post request
+        404 for invalid task_name
+        400 if task can't be started here without argument
+        """
+        # 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)
+
+        if not task_config.get("api_start"):
+            message = {"message": "can not start task through this endpoint"}
+            return Response(message, status=400)
+
+        message = TaskCommand().start(task_name)
+
+        return Response({"message": message})
+
+
+class TaskIDView(ApiBaseView):
+    """resolves to /api/task/by-id/<task-id>/
+    GET: return details of task id
+    POST: send command to task by id
+    """
+
+    valid_commands = ["stop", "kill"]
+    permission_classes = [AdminOnly]
+
+    def get(self, request, task_id):
+        """handle get request"""
+        # 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)
+
+        return Response(task_result)
+
+    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)
+
+        task_result = TaskManager().get_task(task_id)
+        if not task_result:
+            message = {"message": "task id not found"}
+            return Response(message, 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)
+
+            message_key = self._build_message_key(task_conf, task_id)
+            TaskCommand().stop(task_id, message_key)
+        if command == "kill":
+            if not task_conf.get("api_stop"):
+                message = {"message": "task can not be killed"}
+                return Response(message, status=400)
+
+            TaskCommand().kill(task_id)
+
+        return Response({"message": "command sent"})
+
+    def _build_message_key(self, task_conf, task_id):
+        """build message key to forward command to notification"""
+        return f"message:{task_conf.get('group')}:{task_id.split('-')[0]}"
+
+
+class ScheduleView(ApiBaseView):
+    """resolves to /api/task/schedule/<task-name>/
+    DEL: delete schedule for task
+    """
+
+    permission_classes = [AdminOnly]
+
+    def delete(self, request):
+        """delete schedule by task_name query"""
+        task_name = request.data.get("task_name")
+        try:
+            task = CustomPeriodicTask.objects.get(name=task_name)
+        except CustomPeriodicTask.DoesNotExist:
+            message = {"message": "task_name not found"}
+            return Response(message, status=404)
+
+        _ = task.delete()
+
+        return Response({"success": True})
+
+
+class ScheduleNotification(ApiBaseView):
+    """resolves to /api/task/notification/
+    GET: get all schedule notifications
+    DEL: delete notification
+    """
+
+    def get(self, request):
+        """handle get request"""
+
+        return Response(get_all_notifications())
+
+    def delete(self, request):
+        """handle delete"""
+
+        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)
+
+        if url:
+            response, status_code = Notifications(task_name).remove_url(url)
+        else:
+            response, status_code = Notifications(task_name).remove_task()
+
+        return Response({"response": response, "status_code": status_code})