Refactor Task List and filters with error handlers for Scheduler

rocky/katalogus/views/mixins.py

@@ -10,7 +10,6 @@
 from django.utils.translation import gettext_lazy as _
 from requests import HTTPError, RequestException
 from rest_framework.status import HTTP_404_NOT_FOUND
-from tools.view_helpers import schedule_task
 
 from katalogus.client import (
     Boefje as KATalogusBoefje,
@@ -30,6 +29,7 @@
 )
 from rocky.scheduler import Boefje, BoefjeTask, Normalizer, NormalizerTask, QueuePrioritizedItem, RawData
 from rocky.views.mixins import OctopoesView
+from rocky.views.scheduler import schedule_task
 
 logger = getLogger(__name__)
 

rocky/katalogus/views/plugin_detail.py

@@ -16,7 +16,7 @@
 from katalogus.client import get_katalogus
 from katalogus.views.mixins import BoefjeMixin
 from katalogus.views.plugin_settings_list import PluginSettingsListView
-from rocky import scheduler
+from rocky.views.scheduler import get_list_of_tasks_lazy
 
 logger = getLogger(__name__)
 
@@ -52,7 +52,8 @@ def get_task_history(self) -> Page:
 
         page = int(self.request.GET.get("task_history_page", 1))
 
-        task_history = scheduler.client.get_lazy_task_list(
+        task_history = get_list_of_tasks_lazy(
+            self.request,
             scheduler_id=scheduler_id,
             task_type=plugin_type,
             plugin_id=plugin_id,

rocky/rocky/locale/django.pot

@@ -8,7 +8,7 @@ msgid ""
 msgstr ""
 "Project-Id-Version: PACKAGE VERSION\n"
 "Report-Msgid-Bugs-To: \n"
-"POT-Creation-Date: 2023-10-17 13:44+0000\n"
+"POT-Creation-Date: 2023-10-25 14:29+0000\n"
 "PO-Revision-Date: YEAR-MO-DA HO:MI+ZONE\n"
 "Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
 "Language-Team: LANGUAGE <LL@li.org>\n"
@@ -3330,13 +3330,6 @@ msgstr ""
 msgid "Members"
 msgstr ""
 
-#: tools/view_helpers.py
-msgid ""
-"Your task is scheduled and will soon be started in the background. Results "
-"will be added to the object list when they are in. It may take some time, a "
-"refresh of the page may be needed to show the results."
-msgstr ""
-
 #: rocky/messaging.py
 msgid ""
 "You have trusted this member with a clearance level of L{}. This member "
@@ -3345,7 +3338,7 @@ msgid ""
 msgstr ""
 
 #: rocky/scheduler.py
-msgid "Connectivity issues with Mula."
+msgid "Could not connect to Scheduler. Service is possibly down."
 msgstr ""
 
 #: rocky/scheduler.py
@@ -3360,6 +3353,10 @@ msgstr ""
 msgid "Task already queued."
 msgstr ""
 
+#: rocky/scheduler.py
+msgid "Task could not be found."
+msgstr ""
+
 #: rocky/settings.py
 msgid "Blue light"
 msgstr ""
@@ -5524,12 +5521,24 @@ msgstr ""
 msgid "Reset"
 msgstr ""
 
-#: rocky/views/tasks.py
-msgid "Task details not found."
+#: rocky/views/scheduler.py
+msgid "'Plugin not found'"
 msgstr ""
 
-#: rocky/views/tasks.py
-msgid "Fetching tasks failed: no connection with scheduler"
+#: rocky/views/scheduler.py
+msgid "'OOI not found'"
+msgstr ""
+
+#: rocky/views/scheduler.py
+msgid "Scheduling {} {} with input object {} failed. "
+msgstr ""
+
+#: rocky/views/scheduler.py
+msgid ""
+"Task of {} {} with input object {} is scheduled and will soon be started in "
+"the background. Results will be added to the object list when they are in. "
+"It may take some time, a refresh of the page may be needed to show the "
+"results."
 msgstr ""
 
 #: rocky/views/upload_csv.py

rocky/rocky/scheduler.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import datetime
+import logging
 import uuid
 from enum import Enum
 from http import HTTPStatus
@@ -10,10 +11,12 @@
 from django.conf import settings
 from django.utils.translation import gettext_lazy as _
 from pydantic import BaseModel, Field
-from requests.exceptions import HTTPError
+from requests.exceptions import ConnectionError, HTTPError
 
 from rocky.health import ServiceHealth
 
+logger = logging.getLogger(__name__)
+
 
 class Boefje(BaseModel):
     """Boefje representation."""
@@ -164,7 +167,7 @@ def __getitem__(self, key) -> List[Task]:
 
 
 class SchedulerError(Exception):
-    message = _("Connectivity issues with Mula.")
+    message = _("Could not connect to Scheduler. Service is possibly down.")
 
     def __str__(self):
         return str(self.message)
@@ -182,17 +185,25 @@ class ConflictError(SchedulerError):
     message = _("Task already queued.")
 
 
+class TaskNotFoundError(SchedulerError):
+    message = _("Task could not be found.")
+
+
 class SchedulerClient:
     def __init__(self, base_uri: str):
         self.session = requests.Session()
         self._base_uri = base_uri
+        self.is_alive()
 
     def list_tasks(
         self,
         **kwargs,
     ) -> PaginatedTasksResponse:
-        res = self.session.get(f"{self._base_uri}/tasks", params=kwargs)
-        return PaginatedTasksResponse.parse_raw(res.text)
+        try:
+            res = self.session.get(f"{self._base_uri}/tasks", params=kwargs)
+            return PaginatedTasksResponse.parse_raw(res.text)
+        except ConnectionError:
+            raise SchedulerError()
 
     def get_lazy_task_list(
         self,
@@ -205,22 +216,30 @@ def get_lazy_task_list(
         plugin_id: Optional[str] = None,
         boefje_name: Optional[str] = None,
     ) -> LazyTaskList:
-        return LazyTaskList(
-            self,
-            scheduler_id=scheduler_id,
-            type=task_type,
-            status=status,
-            min_created_at=min_created_at,
-            max_created_at=max_created_at,
-            input_ooi=input_ooi,
-            plugin_id=plugin_id,
-            boefje_name=boefje_name,
-        )
+        try:
+            return LazyTaskList(
+                self,
+                scheduler_id=scheduler_id,
+                type=task_type,
+                status=status,
+                min_created_at=min_created_at,
+                max_created_at=max_created_at,
+                input_ooi=input_ooi,
+                plugin_id=plugin_id,
+                boefje_name=boefje_name,
+            )
+        except ConnectionError:
+            raise SchedulerError()
 
     def get_task_details(self, task_id) -> Task:
-        res = self.session.get(f"{self._base_uri}/tasks/{task_id}")
-        res.raise_for_status()
-        return Task.parse_raw(res.content)
+        try:
+            res = self.session.get(f"{self._base_uri}/tasks/{task_id}")
+            res.raise_for_status()
+            return Task.parse_raw(res.content)
+        except HTTPError:
+            raise TaskNotFoundError()
+        except ConnectionError:
+            raise SchedulerError()
 
     def push_task(self, queue_name: str, prioritized_item: QueuePrioritizedItem) -> None:
         try:
@@ -237,10 +256,18 @@ def push_task(self, queue_name: str, prioritized_item: QueuePrioritizedItem) ->
             else:
                 raise SchedulerError()
 
+    def is_alive(self) -> None:
+        try:
+            request = self.session.get(f"{self._base_uri}")
+            request.raise_for_status()
+        except ConnectionError:
+            raise SchedulerError()
+
     def health(self) -> ServiceHealth:
         health_endpoint = self.session.get(f"{self._base_uri}/health")
         health_endpoint.raise_for_status()
         return ServiceHealth.parse_raw(health_endpoint.content)
 
 
-client = SchedulerClient(settings.SCHEDULER_API)
+def get_scheduler() -> SchedulerClient:
+    return SchedulerClient(settings.SCHEDULER_API)

rocky/rocky/views/health.py

@@ -13,7 +13,7 @@
 from rocky.bytes_client import get_bytes_client
 from rocky.health import ServiceHealth
 from rocky.keiko import keiko_client
-from rocky.scheduler import client
+from rocky.scheduler import SchedulerError, get_scheduler
 from rocky.version import __version__
 
 logger = logging.getLogger(__name__)
@@ -54,8 +54,8 @@ def get_octopoes_health(octopoes_api_connector: OctopoesAPIConnector) -> Service
 
 def get_scheduler_health() -> ServiceHealth:
     try:
-        scheduler_health = client.health()
-    except RequestException as ex:
+        scheduler_health = get_scheduler().health()
+    except (SchedulerError, RequestException) as ex:
         logger.exception(ex)
         scheduler_health = ServiceHealth(
             service="scheduler",
@@ -120,6 +120,8 @@ def get_context_data(self, **kwargs):
                 "text": _("Beautified"),
             },
         ]
+
         rocky_health = get_rocky_health(self.octopoes_api_connector)
         context["health_checks"] = flatten_health(rocky_health)
+
         return context

rocky/rocky/views/ooi_detail.py

@@ -21,22 +21,17 @@
 
 from octopoes.models import OOI, Reference
 from octopoes.models.ooi.question import Question
-from rocky import scheduler
 from rocky.views.ooi_detail_related_object import OOIFindingManager, OOIRelatedObjectAddView
 from rocky.views.ooi_view import BaseOOIDetailView
+from rocky.views.scheduler import get_list_of_tasks_lazy
 
 
 class PageActions(Enum):
     START_SCAN = "start_scan"
     SUBMIT_ANSWER = "submit_answer"
 
 
-class OOIDetailView(
-    BoefjeMixin,
-    OOIRelatedObjectAddView,
-    OOIFindingManager,
-    BaseOOIDetailView,
-):
+class OOIDetailView(BoefjeMixin, OOIRelatedObjectAddView, OOIFindingManager, BaseOOIDetailView):
     template_name = "oois/ooi_detail.html"
     connector_form_class = ObservedAtForm
     task_history_limit = 10
@@ -125,7 +120,8 @@ def get_task_history(self) -> Page:
         else:
             max_created_at = None
 
-        task_history = scheduler.client.get_lazy_task_list(
+        task_history = get_list_of_tasks_lazy(
+            self.request,
             scheduler_id=scheduler_id,
             status=status,
             min_created_at=min_created_at,

rocky/rocky/views/scheduler.py

@@ -0,0 +1,69 @@
+import logging
+
+from django.contrib import messages
+from django.http import HttpRequest
+from django.utils.translation import gettext_lazy as _
+
+from rocky.scheduler import (
+    BadRequestError,
+    ConflictError,
+    LazyTaskList,
+    QueuePrioritizedItem,
+    SchedulerError,
+    Task,
+    TaskNotFoundError,
+    TooManyRequestsError,
+    get_scheduler,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def schedule_task(request: HttpRequest, organization_code: str, task: QueuePrioritizedItem) -> None:
+    plugin_name = ""
+    input_ooi = ""
+    plugin_type = task.data.type
+
+    if plugin_type == "boefje":
+        plugin_name = task.data.boefje.name
+        input_ooi = task.data.input_ooi
+    elif plugin_type == "normalizer":
+        plugin_name = task.data.normalizer.id  # name not set yet, is None for name
+        input_ooi = task.data.raw_data.boefje_meta.input_ooi
+    else:
+        plugin_name = _("'Plugin not found'")
+        input_ooi = _("'OOI not found'")
+    try:
+        get_scheduler().push_task(f"{task.data.type}-{organization_code}", task)
+    except (BadRequestError, TooManyRequestsError, ConflictError) as task_error:
+        error_message = (
+            _("Scheduling {} {} with input object {} failed. ").format(plugin_type.title(), plugin_name, input_ooi)
+            + task_error.message
+        )
+        messages.error(request, error_message)
+    except SchedulerError as error:
+        messages.error(request, error.message)
+    else:
+        messages.success(
+            request,
+            _(
+                "Task of {} {} with input object {} is scheduled and will soon be started in the background. "
+                "Results will be added to the object list when they are in. "
+                "It may take some time, a refresh of the page may be needed to show the results."
+            ).format(plugin_type.title(), plugin_name, input_ooi),
+        )
+
+
+def get_list_of_tasks_lazy(request: HttpRequest, **params) -> LazyTaskList:
+    try:
+        return get_scheduler().get_lazy_task_list(**params)
+    except SchedulerError as error:
+        messages.error(request, error.message)
+        return []
+
+
+def get_details_of_task(request: HttpRequest, task_id: str) -> Task:
+    try:
+        return get_scheduler().get_task_details(task_id)
+    except (TaskNotFoundError, TooManyRequestsError, SchedulerError) as error:
+        messages.error(request, error.message)

rocky/rocky/views/task_detail.py

@@ -4,20 +4,16 @@
 from django.views.generic import TemplateView
 from katalogus.views.mixins import BoefjeMixin, NormalizerMixin
 
-from rocky.scheduler import client
 from rocky.views.mixins import OctopoesView
+from rocky.views.scheduler import get_details_of_task
 
 
 class TaskDetailView(OctopoesView, TemplateView):
-    @staticmethod
-    def get_task(task_id):
-        return client.get_task_details(task_id)
-
     def get_context_data(self, **kwargs):
         context = super().get_context_data(**kwargs)
 
         context["task_id"] = kwargs["task_id"]
-        context["task"] = self.get_task(context["task_id"])
+        context["task"] = get_details_of_task(self.request, context["task_id"])
         return context
 
 
@@ -58,7 +54,7 @@ def get_output_oois(self, task):
 
     def get(self, request, *args, **kwargs):
         task_id = kwargs["task_id"]
-        task = self.get_task(task_id)
+        task = get_details_of_task(request, task_id)
         return JsonResponse(
             {
                 "oois": self.get_output_oois(task),