Merge pull request #523 from arabcoders/dev

Feat: Introduce random delays to task handlers

FAQ.md

@@ -4,7 +4,7 @@ Certain configuration values can be set via environment variables, using the `-e
 or the `environment:` section in `compose.yaml` file.
 
 | Environment Variable           | Description                                                         | Default               |
-| ------------------------------ | ------------------------------------------------------------------ | --------------------- |
+| ------------------------------ | ------------------------------------------------------------------- | --------------------- |
 | TZ                             | The timezone to use for the application                             | `(not_set)`           |
 | YTP_OUTPUT_TEMPLATE            | The template for the filenames of the downloaded videos             | `%(title)s.%(ext)s`   |
 | YTP_DEFAULT_PRESET             | The default preset to use for the download                          | `default`             |
@@ -50,6 +50,7 @@ or the `environment:` section in `compose.yaml` file.
 | YTP_STATIC_UI_PATH             | Path to custom static UI files.                                     | `(not_set)`           |
 | YTP_AUTO_CLEAR_HISTORY_DAYS    | Number of days after which completed download history is cleared.   | `0`                   |
 | YTP_DEFAULT_PAGINATION         | The default number of items per page for history.                   | `50`                  |
+| YTP_TASK_HANDLER_RANDOM_DELAY  | The maximum random delay in seconds before starting a task handler. | `60`                  |
 
 > [!NOTE]
 > To raise the maximum workers for specific extractor, you need to add a ENV variable that follows the pattern `YTP_MAX_WORKERS_FOR_<EXTRACTOR_NAME>`.

app/library/Tasks.py

@@ -3,6 +3,7 @@ import inspect
 import json
 import logging
 import pkgutil
+import random
 import time
 import uuid
 from dataclasses import asdict, dataclass, field
@@ -696,6 +697,8 @@ class HandleTask:
     def _dispatcher(self):
         s: dict[list[str]] = {"h": [], "d": [], "u": [], "f": []}
 
+        handler_groups: dict[str, list[tuple[Task, type]]] = {}
+
         for task in self._tasks.get_all():
             if not task.enabled or not task.handler_enabled:
                 s["d"].append(task.name)
@@ -712,19 +715,53 @@ class HandleTask:
                     s["u"].append(task.name)
                     continue
 
-                coro = self.dispatch(task, handler=handler)
+                handler_name = handler.__name__
-                t = asyncio.create_task(coro, name=f"task-{task.id}")
+                if handler_name not in handler_groups:
-                t.add_done_callback(lambda fut, t=task: self._handle_exception(fut, t))
+                    handler_groups[handler_name] = []
+                handler_groups[handler_name].append((task, handler))
                 s["h"].append(task.name)
             except Exception as e:
                 LOG.error(f"Failed to handle task '{task.name}'. '{e!s}'.")
                 s["f"].append(task.name)
 
+        for tasks_with_handlers in handler_groups.values():
+            for idx, (task, handler) in enumerate(tasks_with_handlers):
+                try:
+                    t = asyncio.create_task(
+                        coro=self._dispatch(
+                            task,
+                            handler,
+                            delay=0.0 if 0 == idx else random.uniform(1.0, self._config.task_handler_random_delay),
+                        ),
+                        name=f"task-{task.id}",
+                    )
+                    t.add_done_callback(lambda fut, t=task: self._handle_exception(fut, t))
+                except Exception as e:
+                    LOG.error(f"Failed to dispatch task '{task.name}'. '{e!s}'.")
+
         if len(self._tasks.get_all()) > 0:
             LOG.info(
                 f"Tasks handler summary: Handled: {len(s['h'])}, Unhandled: {len(s['u'])}, Disabled: {len(s['d'])}, Failed: {len(s['f'])}."
             )
 
+    async def _dispatch(self, task: Task, handler: type, delay: float) -> TaskResult | TaskFailure | None:
+        """
+        Dispatch a task after a random delay to avoid rate limiting.
+
+        Args:
+            task (Task): The task to dispatch.
+            handler (type): The handler to use.
+            delay (float): The delay in seconds before dispatching.
+
+        Returns:
+            TaskResult|TaskFailure|None: The dispatch result.
+
+        """
+        if delay > 0:
+            LOG.debug(f"Delaying dispatch of task '{task.name}' by {delay:.1f} seconds.")
+            await asyncio.sleep(delay)
+        return await self.dispatch(task, handler=handler)
+
     def _handle_exception(self, fut: asyncio.Task, task: Task) -> None:
         if fut.cancelled():
             return

app/library/config.py

@@ -201,6 +201,9 @@ class Config(metaclass=Singleton):
     default_pagination: int = 50
     """The default number of items per page for pagination."""
 
+    task_handler_random_delay: float = 60.0
+    """The maximum random delay in seconds before starting a task handler."""
+
     pictures_backends: list[str] = [
         "https://unsplash.it/1920/1080?random",
         "https://picsum.photos/1920/1080",
@@ -265,6 +268,9 @@ class Config(metaclass=Singleton):
     )
     "The variables that are booleans."
 
+    _float_vars: tuple = ("task_handler_random_delay",)
+    "The variables that are floats."
+
     _frontend_vars: tuple = (
         "download_path",
         "keep_archive",
@@ -365,6 +371,9 @@ class Config(metaclass=Singleton):
             if k in self._int_vars:
                 setattr(self, k, int(v))
 
+            if k in self._float_vars:
+                setattr(self, k, float(v))
+
         if isinstance(self.pictures_backends, str) and self.pictures_backends:
             self.pictures_backends = self.pictures_backends.split(",")
 

app/routes/api/images.py

@@ -120,7 +120,7 @@ async def get_background(request: Request, config: Config, cache: Cache) -> Resp
 
     try:
         IS_REQUESTING_BACKGROUND = True
-        backend = random.choice(config.pictures_backends)  # noqa: S311
+        backend = random.choice(config.pictures_backends)
         CACHE_KEY_BING = "random_background_bing"
         CACHE_KEY = "random_background"
 

pyproject.toml

@@ -168,6 +168,7 @@ ignore = [
   "TRY004", # Like it's our choice to use ValuesError :|
   "PT011",
   "RUF001", # We like unicode chars.
+  "S311", # Not used for cryptography
 ]
 
 # Allow fix for all enabled rules (when `--fix`) is provided.