feat(backend): ✨ basic support for background tasks

mealie/services/server_tasks/__init__.py

@@ -0,0 +1,2 @@
+from .background_executory import *
+from .tasks_http_service import *

mealie/services/server_tasks/background_executory.py

@@ -0,0 +1,56 @@
+from random import getrandbits
+from time import sleep
+from typing import Any, Callable
+
+from sqlalchemy.orm import Session
+
+from mealie.db.database import get_database
+from mealie.schema.server.tasks import ServerTask, ServerTaskCreate, ServerTaskNames
+
+from .._base_http_service.http_services import UserHttpService
+
+
+class BackgroundExecutor(UserHttpService):
+    def populate_item(self, _: int) -> ServerTask:
+        pass
+
+    def dispatch(self, task_name: ServerTaskNames, func: Callable, *args: Any, **kwargs: Any) -> ServerTask:
+        """The dispatch function is a wrapper around the BackgroundTasks class in Starlett. It dirctly calls
+        the add_task function and your task will be run in the background. This function all passes the id required
+        to check on the server tasks in the database and provide updates.
+
+        Tasks that are dispachd by the Background executor should be designed to accept this key word argument
+        and update the item in the database accordingly.
+        """
+
+        server_task = ServerTaskCreate(group_id=self.group_id, name=task_name)
+        server_task = self.db.server_tasks.create(server_task)
+
+        self.background_tasks.add_task(func, *args, **kwargs, task_id=server_task.id, session=self.session)
+
+        return server_task
+
+
+def test_executor_func(task_id: int, session: Session) -> None:
+    database = get_database(session)
+    task = database.server_tasks.get_one(task_id)
+
+    task.append_log("test task has started")
+    task.append_log("test task sleeping for 60 seconds")
+
+    sleep(60)
+
+    task.append_log("test task has finished sleep")
+
+    # Randomly Decide to set to failed or not
+
+    is_fail = bool(getrandbits(1))
+
+    if is_fail:
+        task.append_log("test task has failed")
+        task.set_failed()
+    else:
+        task.append_log("test task has succeeded")
+        task.set_finished()
+
+    database.server_tasks.update(task.id, task)

mealie/services/server_tasks/tasks_http_service.py

@@ -0,0 +1,36 @@
+from functools import cached_property
+
+from mealie.schema.server import ServerTask
+from mealie.services._base_http_service.http_services import AdminHttpService, UserHttpService
+
+
+class ServerTasksHttpService(UserHttpService[int, ServerTask]):
+    _restrict_by_group = True
+    _schema = ServerTask
+
+    @cached_property
+    def dal(self):
+        return self.db.server_tasks
+
+    def populate_item(self, id: int) -> ServerTask:
+        self.item = self.dal.get_one(id)
+        return self.item
+
+    def get_all(self) -> list[ServerTask]:
+        return self.dal.multi_query(query_by={"group_id": self.group_id}, order_by="created_at")
+
+
+class AdminServerTasks(AdminHttpService[int, ServerTask]):
+    _restrict_by_group = True
+    _schema = ServerTask
+
+    @cached_property
+    def dal(self):
+        return self.db.server_tasks
+
+    def populate_item(self, id: int) -> ServerTask:
+        self.item = self.dal.get_one(id)
+        return self.item
+
+    def get_all(self) -> list[ServerTask]:
+        return self.dal.get_all(order_by="created_at")