Refactor event handlers to be asynchronous across multiple modules

app/library/DataStore.py

@@ -1,4 +1,3 @@
-import asyncio
 import copy
 import json
 import logging
@@ -151,7 +150,7 @@ class DataStore:
         if "error" == value.info.status and not no_notify:
             from app.library.Events import EventBus, Events
 
-            asyncio.create_task(EventBus.get_instance().emit(Events.ITEM_ERROR, value.info), name="emit_item_error")
+            EventBus.get_instance().emit(Events.ITEM_ERROR, value.info)
 
         self._dict.update({value.info._id: value})
         self._update_store_item(self._type, value.info)

app/library/Download.py

@@ -350,7 +350,7 @@ class Download:
         self.proc.start()
         self.info.status = "preparing"
 
-        await self._notify.emit(Events.ITEM_UPDATED, data=self.info)
+        self._notify.emit(Events.ITEM_UPDATED, data=self.info)
         asyncio.create_task(self.progress_update(), name=f"update-{self.id}")
 
         ret = await asyncio.get_running_loop().run_in_executor(None, self.proc.join)
@@ -516,7 +516,7 @@ class Download:
             self.logger.debug(f"Status Update: {self.info._id=} {status=}")
 
         if isinstance(status, str):
-            await self._notify.emit(Events.ITEM_UPDATED, data=self.info)
+            self._notify.emit(Events.ITEM_UPDATED, data=self.info)
             return
 
         self.tmpfilename: str | None = status.get("tmpfilename")
@@ -547,7 +547,7 @@ class Download:
 
         if "error" == self.info.status and "error" in status:
             self.info.error = status.get("error")
-            await self._notify.emit(
+            self._notify.emit(
                 Events.LOG_ERROR,
                 data=self.info,
                 title="Download Error",
@@ -584,7 +584,7 @@ class Download:
                 self.logger.error(f"Failed to run ffprobe. {status.get}. {e}")
 
         if not self.final_update or fl:
-            await self._notify.emit(Events.ITEM_UPDATED, data=self.info)
+            self._notify.emit(Events.ITEM_UPDATED, data=self.info)
 
     async def progress_update(self):
         """

app/library/DownloadQueue.py

@@ -3,6 +3,7 @@ import functools
 import glob
 import logging
 import time
+import traceback
 import uuid
 from datetime import UTC, datetime, timedelta
 from email.utils import formatdate
@@ -107,9 +108,11 @@ class DownloadQueue(metaclass=Singleton):
             _ (web.Application): The application to attach the download queue to.
 
         """
-        self._notify.subscribe(
+
-            Events.STARTED, lambda _, __: self.initialize(), f"{__class__.__name__}.{__class__.initialize.__name__}"
+        async def event_handler(_, __):
-        )
+            await self.initialize()
+
+        self._notify.subscribe(Events.STARTED, event_handler, f"{__class__.__name__}.{__class__.initialize.__name__}")
 
         Scheduler.get_instance().add(
             timer="* * * * *",
@@ -155,7 +158,6 @@ class DownloadQueue(metaclass=Singleton):
         """
         status: dict[str, str] = {"status": "ok"}
         started = False
-        tasks: list = []
 
         for item_id in ids:
             try:
@@ -173,14 +175,12 @@ class DownloadQueue(metaclass=Singleton):
 
             item.info.auto_start = True
             updated: Download = self.queue.put(item)
-            tasks.append(self._notify.emit(Events.ITEM_UPDATED, data=updated.info))
+            self._notify.emit(Events.ITEM_UPDATED, data=updated.info)
-            tasks.append(
+            self._notify.emit(
-                self._notify.emit(
+                Events.ITEM_RESUMED,
-                    Events.ITEM_RESUMED,
+                data=item.info,
-                    data=item.info,
+                title="Download Resumed",
-                    title="Download Resumed",
+                message=f"Download '{item.info.title}' has been resumed.",
-                    message=f"Download '{item.info.title}' has been resumed.",
-                )
             )
             status[item_id] = "started"
             started = True
@@ -189,9 +189,6 @@ class DownloadQueue(metaclass=Singleton):
         if started:
             self.event.set()
 
-        if len(tasks) > 0:
-            await asyncio.gather(*tasks)
-
         return status
 
     async def pause_items(self, ids: list[str]) -> dict[str, str]:
@@ -206,7 +203,6 @@ class DownloadQueue(metaclass=Singleton):
 
         """
         status: dict[str, str] = {"status": "ok"}
-        tasks: list = []
 
         for item_id in ids:
             try:
@@ -229,21 +225,16 @@ class DownloadQueue(metaclass=Singleton):
 
             item.info.auto_start = False
             updated: Download = self.queue.put(item)
-            tasks.append(self._notify.emit(Events.ITEM_UPDATED, data=updated.info))
+            self._notify.emit(Events.ITEM_UPDATED, data=updated.info)
-            tasks.append(
+            self._notify.emit(
-                self._notify.emit(
+                Events.ITEM_PAUSED,
-                    Events.ITEM_PAUSED,
+                data=item.info,
-                    data=item.info,
+                title="Download Paused",
-                    title="Download Paused",
+                message=f"Download '{item.info.title}' has been paused.",
-                    message=f"Download '{item.info.title}' has been paused.",
-                )
             )
             status[item_id] = "paused"
             LOG.debug(f"Item {item.info.name()} marked as paused.")
 
-        if len(tasks) > 0:
-            await asyncio.gather(*tasks)
-
         return status
 
     def pause(self, shutdown: bool = False) -> bool:
@@ -495,7 +486,7 @@ class DownloadQueue(metaclass=Singleton):
 
                 dlInfo.info.status = "not_live"
                 dlInfo.info.msg = nMessage.replace(f" '{dlInfo.info.title}'", "")
-                await self._notify.emit(
+                self._notify.emit(
                     Events.LOG_INFO,
                     data={"preset": dlInfo.info.preset, "lowPriority": True},
                     title=nTitle,
@@ -517,7 +508,7 @@ class DownloadQueue(metaclass=Singleton):
                 dlInfo.info.status = "error"
                 itemDownload = self.done.put(dlInfo)
 
-                await self._notify.emit(
+                self._notify.emit(
                     Events.LOG_WARNING,
                     data={"preset": dlInfo.info.preset, "logs": text_logs},
                     title=nTitle,
@@ -557,7 +548,7 @@ class DownloadQueue(metaclass=Singleton):
                     nStore = "history"
                     nEvent = Events.ITEM_MOVED
                     nTitle = "Premiering right now"
-                    await self._notify.emit(
+                    self._notify.emit(
                         Events.LOG_INFO, data={"preset": dlInfo.info.preset}, title=nTitle, message=nMessage
                     )
             else:
@@ -570,7 +561,7 @@ class DownloadQueue(metaclass=Singleton):
                 else:
                     LOG.debug(f"Item {itemDownload.info.name()} is not set to auto-start.")
 
-            await self._notify.emit(
+            self._notify.emit(
                 nEvent,
                 data={"to": nStore, "preset": itemDownload.info.preset, "item": itemDownload.info}
                 if Events.ITEM_MOVED == nEvent
@@ -702,7 +693,7 @@ class DownloadQueue(metaclass=Singleton):
 
                         self.done.put(dlInfo)
 
-                        await self._notify.emit(
+                        self._notify.emit(
                             Events.ITEM_MOVED,
                             data={"to": "history", "preset": dlInfo.info.preset, "item": dlInfo.info},
                             title="Download History Update",
@@ -712,7 +703,7 @@ class DownloadQueue(metaclass=Singleton):
 
                 message: str = f"The URL '{item.url}' is already downloaded and recorded in archive."
                 LOG.error(message)
-                await self._notify.emit(
+                self._notify.emit(
                     Events.LOG_INFO, data={"preset": item.preset}, title="Already Downloaded", message=message
                 )
 
@@ -866,7 +857,7 @@ class DownloadQueue(metaclass=Singleton):
                 await item.close()
                 LOG.debug(f"Deleting from queue {item_ref}")
                 self.queue.delete(id)
-                await self._notify.emit(
+                self._notify.emit(
                     Events.ITEM_CANCELLED,
                     data=item.info,
                     title="Download Cancelled",
@@ -874,7 +865,7 @@ class DownloadQueue(metaclass=Singleton):
                 )
                 item.info.status = "cancelled"
                 self.done.put(item)
-                await self._notify.emit(
+                self._notify.emit(
                     Events.ITEM_MOVED,
                     data={"to": "history", "preset": item.info.preset, "item": item.info},
                     title="Download Cancelled",
@@ -949,7 +940,7 @@ class DownloadQueue(metaclass=Singleton):
             self.done.delete(id)
 
             _status: str = "Removed" if removed_files > 0 else "Cleared"
-            await self._notify.emit(
+            self._notify.emit(
                 Events.ITEM_DELETED,
                 data=item.info,
                 title=f"Download {_status}",
@@ -1086,7 +1077,6 @@ class DownloadQueue(metaclass=Singleton):
             await entry.close()
 
         if self.queue.exists(key=id):
-            _tasks = []
             LOG.debug(f"Download Task '{id}' is completed. Removing from queue.")
             self.queue.delete(key=id)
 
@@ -1096,7 +1086,7 @@ class DownloadQueue(metaclass=Singleton):
             if entry.is_cancelled() is True:
                 nTitle = "Download Cancelled"
                 nMessage = f"Cancelled '{entry.info.title}' download."
-                await self._notify.emit(Events.ITEM_CANCELLED, data=entry.info, title=nTitle, message=nMessage)
+                self._notify.emit(Events.ITEM_CANCELLED, data=entry.info, title=nTitle, message=nMessage)
                 entry.info.status = "cancelled"
 
             if entry.info.status == "finished" and entry.info.filename:
@@ -1105,19 +1095,15 @@ class DownloadQueue(metaclass=Singleton):
                 if entry.info.is_archivable and not entry.info.is_archived:
                     entry.info.is_archived = True
 
-                _tasks.append(self._notify.emit(Events.ITEM_COMPLETED, data=entry.info, title=nTitle, message=nMessage))
+                self._notify.emit(Events.ITEM_COMPLETED, data=entry.info, title=nTitle, message=nMessage)
 
             self.done.put(entry)
-            _tasks.append(
+            self._notify.emit(
-                self._notify.emit(
+                Events.ITEM_MOVED,
-                    Events.ITEM_MOVED,
+                data={"to": "history", "preset": entry.info.preset, "item": entry.info},
-                    data={"to": "history", "preset": entry.info.preset, "item": entry.info},
+                title=nTitle,
-                    title=nTitle,
+                message=nMessage,
-                    message=nMessage,
-                )
             )
-
-            await asyncio.gather(*_tasks)
         else:
             LOG.warning(f"Download '{id}' not found in queue.")
 
@@ -1223,7 +1209,6 @@ class DownloadQueue(metaclass=Singleton):
             return
 
         if exc := task.exception():
-            import traceback
 
             task_name: str = task.get_name() if task.get_name() else "unknown_task"
             LOG.error(f"Unhandled exception in background task '{task_name}': {exc!s}. {traceback.format_exc()}")

app/library/Events.py

@@ -145,6 +145,9 @@ class Event:
     data: Any
     """The data that was passed to the event."""
 
+    extras: dict = field(default_factory=dict)
+    """Listeners can add extra data to the event."""
+
     def serialize(self) -> dict:
         """
         Serialize the event.
@@ -162,9 +165,20 @@ class Event:
             "data": self.data,
         }
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         return f"Event(id={self.id}, created_at={self.created_at}, event={self.event}, title={self.title}, message={self.message} data={self.data})"
 
+    def put(self, key: str, value: Any) -> None:
+        """
+        Put extra data to the event.
+
+        Args:
+            key (str): The key of the extra data.
+            value (Any): The value of the extra data.
+
+        """
+        self.extras[key] = value
+
     def datatype(self) -> str:
         """
         Get the datatype of the data.
@@ -210,17 +224,12 @@ class EventBus(metaclass=Singleton):
     debug: bool = False
     """Whether to log debug messages or not."""
 
-    _offload: BackgroundWorker
+    _offload: BackgroundWorker = None
     """The background worker to offload tasks to."""
 
     def __init__(self):
         EventBus._instance = self
 
-        from .config import Config
-
-        self.debug = Config.get_instance().debug
-        self._offload = BackgroundWorker.get_instance()
-
     @staticmethod
     def get_instance() -> "EventBus":
         """
@@ -305,84 +314,11 @@ class EventBus(metaclass=Singleton):
 
         return self
 
-    def sync_emit(
+    def emit(
-        self,
-        event: str,
-        data: Any | None = None,
-        title: str | None = None,
-        message: str | None = None,
-        loop=None,
-        wait: bool = True,
-        **kwargs,
-    ):
-        """
-        Emit event and (optionally) wait for results.
-
-        Args:
-            event (str): The event to emit.
-            data (Any|None): The data to pass to the event.
-            title (str | None): The title of the event, if any.
-            message (str | None): The message of the event, if any.
-            loop (asyncio.AbstractEventLoop | None): The event loop to use. If None, the current running loop is used.
-            wait (bool): Whether to wait for the results of the event handlers. Defaults to True.
-            **kwargs: Additional keyword arguments to pass to the event
-
-        Returns:
-            list: The results are the return values of the coroutines. If the coroutine raises an exception,
-                  the exception is caught and logged. If event does not exist, an empty list is returned.
-                  If wait is False, a list of asyncio.Task objects is returned instead if we are in a running event loop,
-                  or the result of the coroutine if we are not in a running event loop.
-
-        """
-        if event not in self._listeners:
-            return []
-
-        if not data:
-            data = {}
-
-        async def emit_all():
-            ev = Event(event=event, title=title, message=message, data=data)
-            LOG.debug(f"Emitting event '{ev.id}: {ev.event}'.", extra={"data": data})
-
-            res: list = []
-
-            for h in self._listeners[event].values():
-                try:
-                    res.append(await h.handle(ev, **kwargs))
-                except Exception as e:
-                    LOG.exception(e)
-                    LOG.error(f"Failed to emit event '{event}' to '{h.name}'. Error message '{e!s}'.")
-
-            return res
-
-        try:
-            loop = loop or asyncio.get_running_loop()
-            in_same_loop: bool = asyncio.get_running_loop() is loop
-        except RuntimeError:
-            loop = None
-            in_same_loop = False
-
-        if loop is None or not loop.is_running():
-            return asyncio.run(emit_all())
-
-        if in_same_loop:
-            if wait:
-                msg = (
-                    "Calling EventsBus.sync_emit(...,wait=True) from within the running event loop would cause dead-lock. "
-                    "Use `await EventsBus.emit(...)` or `EventsBus.sync_emit(..., wait=False)`."
-                )
-                raise RuntimeError(msg)
-
-            return loop.create_task(emit_all())
-
-        fut = asyncio.run_coroutine_threadsafe(emit_all(), loop)
-        return fut.result() if wait else fut
-
-    async def emit(
         self, event: str, data: Any | None = None, title: str | None = None, message: str | None = None, **kwargs
-    ) -> Awaitable:
+    ) -> None:
         """
-        Emit an event.
+        Emit an event to all registered listeners.
 
         Args:
             event (str): The event to emit.
@@ -391,44 +327,6 @@ class EventBus(metaclass=Singleton):
             message (str | None): The message of the event, if any.
             **kwargs: The keyword arguments to pass to the event.
 
-        Returns:
-            Awaitable: The task that was created to run the event.
-
-        """
-        if event not in self._listeners:
-            return []
-
-        if not data:
-            data = {}
-
-        ev = Event(event=event, title=title, message=message, data=data)
-
-        if self.debug or event not in Events.only_debug():
-            LOG.debug(f"Emitting event '{ev.id}: {ev.event}'.", extra={"data": data})
-
-        tasks = []
-        for handler in self._listeners[event].values():
-            try:
-                tasks.append(handler.handle(ev, **kwargs))
-            except Exception as e:
-                LOG.exception(e)
-                LOG.error(f"Failed to emit event '{ev.event}' to '{handler.name}'. Error message '{e!s}'.")
-
-        return asyncio.gather(*tasks)
-
-    def offload(
-        self, event: str, title: str | None = None, message: str | None = None, data: Any | None = None, **kwargs
-    ) -> None:
-        """
-        Offload an dispatching event to a background worker.
-
-        Args:
-            event (str): The event to offload.
-            data (Any|None): The data to pass to the event.
-            title (str | None): The title of the event, if any.
-            message (str | None): The message of the event, if any.
-            **kwargs: Additional keyword arguments to pass to the event.
-
         """
         if event not in self._listeners:
             return
@@ -439,11 +337,57 @@ class EventBus(metaclass=Singleton):
         ev = Event(event=event, title=title, message=message, data=data)
 
         if self.debug or event not in Events.only_debug():
-            LOG.debug(f"Offloading event '{ev.id}: {ev.event}'.", extra={"data": data})
+            LOG.debug(f"Emitting event '{ev.id}: {ev.event}'.", extra={"data": data})
 
-        for handler in self._listeners[event].values():
+        try:
-            try:
+            loop = asyncio.get_running_loop()
-                self._offload.submit(handler.handle, ev, **kwargs)
+
-            except Exception as e:
+            for handler in self._listeners[event].values():
-                LOG.exception(e)
+                try:
-                LOG.error(f"Failed to offload event '{ev.event}' to '{handler.name}'. Error message '{e!s}'.")
+                    if handler.is_coroutine:
+                        coro = handler.call_back(ev, handler.name, **kwargs)
+                        if asyncio.iscoroutine(coro):
+                            loop.create_task(coro)
+                        else:
+                            LOG.warning(f"Expected coroutine from async handler '{handler.name}', got {type(coro)}")
+                    else:
+                        loop.create_task(self._call(handler, ev, kwargs), name=f"sync-handler-{handler.name}-{ev.id}")
+                except Exception as e:
+                    LOG.exception(e)
+                    LOG.error(f"Failed to emit event '{ev.event}' to '{handler.name}'. Error message '{e!s}'.")
+        except RuntimeError:
+            LOG.debug(f"No event loop detected - using BackgroundWorker for {len(self._listeners[event])} handlers")
+            for handler in self._listeners[event].values():
+                try:
+                    if not self._offload:
+                        self._offload = BackgroundWorker.get_instance()
+
+                    self._offload.submit(handler.handle, ev, **kwargs)
+                except Exception as e:
+                    LOG.exception(e)
+                    LOG.error(f"Failed to emit event '{ev.event}' to '{handler.name}'. Error message '{e!s}'.")
+
+    def clear(self) -> None:
+        """
+        Clear all listeners. Useful for testing.
+        """
+        self._listeners.clear()
+
+    def debug_enable(self) -> None:
+        """
+        Enable debug logging.
+        """
+        self.debug = True
+
+    def debug_disable(self) -> None:
+        """
+        Disable debug logging.
+        """
+        self.debug = False
+
+    @staticmethod
+    def _call(h, event, kw):
+        async def call_handler():
+            return h.call_back(event, h.name, **kw)
+
+        return call_handler()

app/library/HttpSocket.py

@@ -52,10 +52,10 @@ class HttpSocket:
             ping_timeout=5,
         )
         encoder = encoder or Encoder()
-        self.rootPath = root_path
+        self.rootPath: Path = root_path
 
-        def emit(e: Event, _, **kwargs):
+        async def event_handler(e: Event, _, **kwargs):
-            return self.sio.emit(event=e.event, data=encoder.encode(e), **kwargs)
+            await self.sio.emit(event=e.event, data=encoder.encode(e), **kwargs)
 
         services = Services.get_instance()
         services.add_all(
@@ -73,7 +73,7 @@ class HttpSocket:
             }
         )
 
-        self._notify.subscribe("frontend", emit, f"{__class__.__name__}.emit")
+        self._notify.subscribe("frontend", event_handler, f"{__class__.__name__}.emit")
 
     @staticmethod
     def ws_event(func):  # type: ignore
@@ -101,11 +101,12 @@ class HttpSocket:
         app.on_shutdown.append(self.on_shutdown)
 
         self.sio.attach(app, socketio_path=f"{self.config.base_path.rstrip('/')}/socket.io")
-        self._notify.subscribe(
+
-            Events.ADD_URL,
+        async def event_handler(data: Event, _):
-            lambda data, _, **kwargs: self.queue.add(item=Item.format(data.data)),  # noqa: ARG005
+            if data and data.data:
-            f"{__class__.__name__}.add",
+                await self.queue.add(item=Item.format(data.data))
-        )
+
+        self._notify.subscribe(Events.ADD_URL, event_handler, f"{__class__.__name__}.add")
 
         load_modules(self.rootPath, self.rootPath / "routes" / "socket")
 

app/library/Notifications.py

@@ -505,13 +505,12 @@ class Notification(metaclass=Singleton):
             LOG.error(f"Error sending Notification event '{ev.event}: {ev.id}' to '{target.name}'. '{err_msg!s}'.")
             return {"url": target.request.url, "status": 500, "text": str(ev)}
 
-    def emit(self, e: Event, _, **__):
+    def emit(self, e: Event, _, **__) -> None:
         if len(self._targets) < 1 or not NotificationEvents.is_valid(e.event):
-            return self.noop()
+            return
 
         self._offload.submit(self.send, e)
-
+        return
-        return self.noop()
 
     def _deep_unpack(self, data: dict) -> dict:
         for k, v in data.items():

app/library/Presets.py

@@ -136,7 +136,7 @@ class Presets(metaclass=Singleton):
                 LOG.error(f"Failed to parse default preset ':{i}'. '{e!s}'.")
                 continue
 
-        def event_handler(_, __):
+        async def event_handler(_, __):
             msg = "Not implemented"
             raise Exception(msg)
 

app/library/Scheduler.py

@@ -26,11 +26,11 @@ class Scheduler(metaclass=Singleton):
 
         self._loop = loop or asyncio.get_event_loop()
 
-        EventBus.get_instance().subscribe(
+        async def event_handler(data, _):
-            Events.SCHEDULE_ADD,
+            if data and data.data:
-            lambda data, _, **kwargs: self.add(**data.data),  # noqa: ARG005
+                self.add(**data.data)
-            f"{__class__.__name__}.add",
+
-        )
+        EventBus.get_instance().subscribe(Events.SCHEDULE_ADD, event_handler, f"{__class__.__name__}.add")
 
     @staticmethod
     def get_instance() -> "Scheduler":

app/library/Tasks.py

@@ -193,11 +193,12 @@ class Tasks(metaclass=Singleton):
 
         """
         self.load()
-        self._notify.subscribe(
+
-            Events.TASKS_ADD,
+        async def event_handler(data, _):
-            lambda data, _, **kwargs: self.save(**data.data),  # noqa: ARG005
+            if data and data.data:
-            f"{__class__.__name__}.add",
+                self.save(data.data)
-        )
+
+        self._notify.subscribe(Events.TASKS_ADD, event_handler, f"{__class__.__name__}.add")
         self._task_handler.load()
 
     def get_all(self) -> list[Task]:
@@ -449,7 +450,7 @@ class Tasks(metaclass=Singleton):
             await asyncio.gather(*_tasks)
         except Exception as e:
             LOG.error(f"Failed to execute '{task.name}' at '{timeNow}'. '{e!s}'.")
-            await self._notify.emit(
+            self._notify.emit(
                 Events.LOG_ERROR,
                 data={"preset": task.preset},
                 title="Task failed",

app/library/conditions.py

@@ -68,7 +68,7 @@ class Conditions(metaclass=Singleton):
             except Exception:
                 pass
 
-        def event_handler(_, __):
+        async def event_handler(_, __):
             msg = "Not implemented"
             raise Exception(msg)
 

app/library/dl_fields.py

@@ -122,7 +122,7 @@ class DLFields(metaclass=Singleton):
             except Exception:
                 pass
 
-        def event_handler(_, __):
+        async def event_handler(_, __):
             msg = "Not implemented"
             raise Exception(msg)
 

app/library/task_handlers/_base_handler.py

@@ -26,11 +26,11 @@ class BaseHandler:
         if "failure_count" not in cls.__dict__:
             cls.failure_count = {}
 
-        EventBus.get_instance().subscribe(
+        async def event_handler(data, _):
-            Events.ITEM_ERROR,
+            if data and data.data:
-            lambda data, _, **__: cls.on_error(data.data),
+                await cls.on_error(data.data)
-            f"{cls.__name__}.on_error",
+
-        )
+        EventBus.get_instance().subscribe(Events.ITEM_ERROR, event_handler, f"{cls.__name__}.on_error")
 
     @staticmethod
     def can_handle(task: Task) -> bool:

app/library/task_handlers/twitch.py

@@ -1,4 +1,3 @@
-import asyncio
 import logging
 import re
 from typing import TYPE_CHECKING
@@ -146,9 +145,8 @@ class TwitchHandler(BaseHandler):
         )
 
         try:
-            await asyncio.gather(
+            for item in filtered:
-                *[notify.emit(Events.ADD_URL, data=rItem.new_with(url=item["url"]).serialize()) for item in filtered]
+                notify.emit(Events.ADD_URL, data=rItem.new_with(url=item["url"]).serialize())
-            )
         except Exception as e:
             LOG.exception(e)
             LOG.error(f"Error while adding items from '{task.name}'. {e!s}")

app/library/task_handlers/youtube.py

@@ -1,4 +1,3 @@
-import asyncio
 import logging
 import re
 from typing import TYPE_CHECKING
@@ -153,9 +152,8 @@ class YoutubeHandler(BaseHandler):
         )
 
         try:
-            await asyncio.gather(
+            for item in filtered:
-                *[notify.emit(Events.ADD_URL, data=rItem.new_with(url=item["url"]).serialize()) for item in filtered]
+                notify.emit(Events.ADD_URL, data=rItem.new_with(url=item["url"]).serialize())
-            )
         except Exception as e:
             LOG.exception(e)
             LOG.error(f"'{task.name}': Error while adding items from task feed. {e!s}")

app/main.py

@@ -38,7 +38,7 @@ ROOT_PATH: Path = Path(__file__).parent.absolute()
 
 class Main:
     def __init__(self, is_native: bool = False):
-        self._config = Config.get_instance(is_native=is_native)
+        self._config: Config = Config.get_instance(is_native=is_native)
         self._app = web.Application()
         self._app.on_shutdown.append(self.on_shutdown)
         self._background_worker = BackgroundWorker()
@@ -98,7 +98,7 @@ class Main:
             raise
 
     async def on_shutdown(self, _: web.Application):
-        await EventBus.get_instance().emit(
+        EventBus.get_instance().emit(
             Events.SHUTDOWN,
             data={"app": self._app},
             title="Application Shutdown",
@@ -112,12 +112,15 @@ class Main:
         host = host or self._config.host
         port = port or self._config.port
 
-        EventBus.get_instance().sync_emit(
+        EventBus.get_instance().emit(
             Events.STARTUP,
             data={"app": self._app},
             title="Application Startup",
             message="The application is starting up.",
         )
+        if self._config.debug:
+            EventBus.get_instance().debug_enable()
+
         Scheduler.get_instance().attach(self._app)
 
         self._socket.attach(self._app)
@@ -131,7 +134,7 @@ class Main:
         DLFields.get_instance().attach(self._app)
         self._background_worker.attach(self._app)
 
-        EventBus.get_instance().sync_emit(
+        EventBus.get_instance().emit(
             Events.LOADED,
             data={"app": self._app},
             title="Application Loaded",
@@ -147,13 +150,11 @@ class Main:
 
             loop = asyncio.get_event_loop()
 
-            EventBus.get_instance().sync_emit(
+            EventBus.get_instance().emit(
                 Events.STARTED,
                 data={"app": self._app},
                 title="Application Started",
                 message="The application has started successfully.",
-                loop=loop,
-                wait=False,
             )
 
             if loop and self._config.debug:

app/routes/api/conditions.py

@@ -114,7 +114,7 @@ async def conditions_add(request: Request, encoder: Encoder, notify: EventBus) -
             status=web.HTTPInternalServerError.status_code,
         )
 
-    await notify.emit(Events.CONDITIONS_UPDATE, data=items)
+    notify.emit(Events.CONDITIONS_UPDATE, data=items)
     return web.json_response(data=items, status=web.HTTPOk.status_code, dumps=encoder.encode)
 
 

app/routes/api/dl_fields.py

@@ -96,5 +96,5 @@ async def dl_fields_add(request: Request, encoder: Encoder, notify: EventBus) ->
             status=web.HTTPInternalServerError.status_code,
         )
 
-    await notify.emit(Events.DLFIELDS_UPDATE, data=items)
+    notify.emit(Events.DLFIELDS_UPDATE, data=items)
     return web.json_response(data=items, status=web.HTTPOk.status_code, dumps=encoder.encode)

app/routes/api/history.py

@@ -169,7 +169,7 @@ async def item_update(request: Request, queue: DownloadQueue, encoder: Encoder,
 
     if updated:
         queue.done.put(item)
-        await notify.emit(Events.ITEM_UPDATED, data=item.info)
+        notify.emit(Events.ITEM_UPDATED, data=item.info)
 
     return web.json_response(
         data=item.info,
@@ -309,7 +309,7 @@ async def item_archive_add(request: Request, queue: DownloadQueue, notify: Event
 
     item.info.archive_status(force=True)
     queue.done.put(item, no_notify=True)
-    await notify.emit(Events.ITEM_UPDATED, data=item.info)
+    notify.emit(Events.ITEM_UPDATED, data=item.info)
 
     return web.json_response(
         data={"message": f"item '{item.info.title}' archived."},
@@ -367,7 +367,7 @@ async def item_archive_delete(request: Request, queue: DownloadQueue, notify: Ev
 
     item.info.archive_status(force=True)
     queue.done.put(item, no_notify=True)
-    await notify.emit(Events.ITEM_UPDATED, data=item.info)
+    notify.emit(Events.ITEM_UPDATED, data=item.info)
 
     return web.json_response(
         data={"message": f"item '{item.info.title}' removed from archive."},

app/routes/api/notifications.py

@@ -104,6 +104,6 @@ async def notification_test(encoder: Encoder, notify: EventBus) -> Response:
         Response: The response object.
 
     """
-    await notify.emit(Events.TEST, title="Test Notification", message="This is a test notification.")
+    notify.emit(Events.TEST, title="Test Notification", message="This is a test notification.")
 
     return web.json_response(data={}, status=web.HTTPOk.status_code, dumps=encoder.encode)

app/routes/api/presets.py

@@ -96,5 +96,5 @@ async def presets_add(request: Request, encoder: Encoder, notify: EventBus) -> R
             status=web.HTTPInternalServerError.status_code,
         )
 
-    await notify.emit(Events.PRESETS_UPDATE, data=presets)
+    notify.emit(Events.PRESETS_UPDATE, data=presets)
     return web.json_response(data=presets, status=web.HTTPOk.status_code, dumps=encoder.encode)

app/routes/api/system.py

@@ -40,7 +40,7 @@ async def downloads_pause(queue: DownloadQueue, encoder: Encoder, notify: EventB
     queue.pause()
 
     msg = "Non-active downloads have been paused."
-    await notify.emit(
+    notify.emit(
         Events.PAUSED,
         data={"paused": True, "at": time.time()},
         title="Downloads Paused",
@@ -75,7 +75,7 @@ async def downloads_resume(queue: DownloadQueue, encoder: Encoder, notify: Event
     queue.resume()
 
     msg = "Resumed all downloads."
-    await notify.emit(
+    notify.emit(
         Events.RESUMED,
         data={"paused": False, "at": time.time()},
         title="Downloads Resumed",
@@ -109,7 +109,7 @@ async def shutdown_system(request: Request, config: Config, encoder: Encoder, no
     app = request.app
 
     async def do_shutdown():
-        await notify.emit(
+        notify.emit(
             Events.SHUTDOWN,
             data={"app": app},
             title="Application Shutdown",

app/routes/socket/connection.py

@@ -37,7 +37,7 @@ async def connect(config: Config, queue: DownloadQueue, notify: EventBus, sid: s
         depth_limit=config.download_path_depth-1,
     )
 
-    await notify.emit(
+    notify.emit(
         Events.CONNECTED,
         data=data,
         title="Client connected",
@@ -78,7 +78,7 @@ async def subscribe(config: Config, notify: EventBus, sio: socketio.AsyncServer,
 
     """
     if not isinstance(data, str) or not data:
-        await notify.emit(
+        notify.emit(
             Events.LOG_ERROR,
             title="Subscription Error",
             message="Invalid event type was expecting a string.",

app/routes/socket/history.py

@@ -12,13 +12,13 @@ LOG: logging.Logger = logging.getLogger(__name__)
 async def add_url(queue: DownloadQueue, notify: EventBus, sid: str, data: dict):
     data = data if isinstance(data, dict) else {}
     if not (url := data.get("url", None)):
-        await notify.emit(Events.LOG_ERROR, title="Invalid request", message="No URL provided.", to=sid)
+        notify.emit(Events.LOG_ERROR, title="Invalid request", message="No URL provided.", to=sid)
         return
 
     item: Item = Item.format(data)
     try:
         status = await queue.add(item=item)
-        await notify.emit(
+        notify.emit(
             event=Events.ITEM_STATUS,
             title="Adding URL",
             message=f"Adding URL '{url}' to the download queue.",
@@ -27,7 +27,7 @@ async def add_url(queue: DownloadQueue, notify: EventBus, sid: str, data: dict):
         )
     except ValueError as e:
         LOG.exception(e)
-        await notify.emit(
+        notify.emit(
             Events.LOG_ERROR, data={"preset": item.preset}, title="Error Adding URL", message=str(e), to=sid
         )
 
@@ -35,7 +35,7 @@ async def add_url(queue: DownloadQueue, notify: EventBus, sid: str, data: dict):
 @route(RouteType.SOCKET, "item_cancel", "item_cancel")
 async def item_cancel(queue: DownloadQueue, notify: EventBus, sid: str, data: str):
     if not (data := data if isinstance(data, str) else None):
-        await notify.emit(Events.LOG_ERROR, title="Invalid Request", message="No item ID provided.", to=sid)
+        notify.emit(Events.LOG_ERROR, title="Invalid Request", message="No item ID provided.", to=sid)
         return
 
     await queue.cancel([data])
@@ -46,7 +46,7 @@ async def item_delete(queue: DownloadQueue, notify: EventBus, sid: str, data: di
     data = data if isinstance(data, dict) else {}
 
     if not (id := data.get("id", None)):
-        await notify.emit(Events.LOG_ERROR, title="Invalid Request", message="No item ID provided.", to=sid)
+        notify.emit(Events.LOG_ERROR, title="Invalid Request", message="No item ID provided.", to=sid)
         return
 
     await queue.clear([id], remove_file=bool(data.get("remove_file", False)))
@@ -55,7 +55,7 @@ async def item_delete(queue: DownloadQueue, notify: EventBus, sid: str, data: di
 @route(RouteType.SOCKET, "item_start", "item_start")
 async def item_start(queue: DownloadQueue, notify: EventBus, sid: str, data: list | str) -> None:
     if not data:
-        await notify.emit(
+        notify.emit(
             Events.LOG_ERROR,
             title="Invalid Request",
             message="No items provided to start.",
@@ -72,7 +72,7 @@ async def item_start(queue: DownloadQueue, notify: EventBus, sid: str, data: lis
 @route(RouteType.SOCKET, "item_pause", "item_pause")
 async def item_pause(queue: DownloadQueue, notify: EventBus, sid: str, data: list | str) -> None:
     if not data:
-        await notify.emit(
+        notify.emit(
             Events.LOG_ERROR,
             title="Invalid Request",
             message="No items provided to pause.",

app/routes/socket/terminal.py

@@ -17,7 +17,7 @@ LOG: logging.Logger = logging.getLogger(__name__)
 @route(RouteType.SOCKET, "cli_post", "socket_cli_post")
 async def cli_post(config: Config, notify: EventBus, sid: str, data: str):
     if not config.console_enabled:
-        await notify.emit(
+        notify.emit(
             Events.LOG_ERROR,
             title="Feature disabled",
             message="Console feature is disabled.",
@@ -26,7 +26,7 @@ async def cli_post(config: Config, notify: EventBus, sid: str, data: str):
         return
 
     if not data:
-        await notify.emit(Events.CLI_CLOSE, data={"exitcode": 0}, to=sid)
+        notify.emit(Events.CLI_CLOSE, data={"exitcode": 0}, to=sid)
         return
 
     import asyncio
@@ -93,7 +93,7 @@ async def cli_post(config: Config, notify: EventBus, sid: str, data: str):
                 assert proc.stdout is not None
                 async for raw_line in proc.stdout:
                     line = raw_line.rstrip(b"\n")
-                    await notify.emit(
+                    notify.emit(
                         Events.CLI_OUTPUT,
                         data={"type": "stdout", "line": line.decode("utf-8", errors="replace")},
                         to=sid,
@@ -112,7 +112,7 @@ async def cli_post(config: Config, notify: EventBus, sid: str, data: str):
 
                 if not chunk:
                     if buffer:
-                        await notify.emit(
+                        notify.emit(
                             Events.CLI_OUTPUT,
                             data={"type": "stdout", "line": buffer.decode("utf-8", errors="replace")},
                             to=sid,
@@ -123,7 +123,7 @@ async def cli_post(config: Config, notify: EventBus, sid: str, data: str):
                 *lines, buffer = buffer.split(b"\n")
 
                 for line in lines:
-                    await notify.emit(
+                    notify.emit(
                         Events.CLI_OUTPUT,
                         data={"type": "stdout", "line": line.decode("utf-8", errors="replace")},
                         to=sid,
@@ -141,6 +141,6 @@ async def cli_post(config: Config, notify: EventBus, sid: str, data: str):
     except Exception as e:
         LOG.error(f"CLI execute exception was thrown for client '{sid}'.")
         LOG.exception(e)
-        await notify.emit(Events.CLI_OUTPUT, data={"type": "stderr", "line": str(e)}, to=sid)
+        notify.emit(Events.CLI_OUTPUT, data={"type": "stderr", "line": str(e)}, to=sid)
     finally:
-        await notify.emit(Events.CLI_CLOSE, data={"exitcode": returncode}, to=sid)
+        notify.emit(Events.CLI_CLOSE, data={"exitcode": returncode}, to=sid)

pyproject.toml

@@ -90,6 +90,9 @@ installer = ["pyinstaller"]
 select = [
   "ALL", # include all the rules, including new ones
 ]
+# Ignore some rules in test files.
+per-file-ignores = { "test_*.py" = ["D"] }
+
 ignore = [
   #### modules
   "ANN", # flake8-annotations
@@ -206,7 +209,4 @@ testpaths = ["app/tests"]
 addopts = "-v --tb=short"
 
 [dependency-groups]
-dev = [
+dev = ["pytest>=8.4.2", "pytest-asyncio>=1.1.0", "ruff>=0.13.0"]
-    "pytest>=8.4.2",
-    "ruff>=0.13.0",
-]

uv.lock

@@ -1041,6 +1041,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/a8/a4/20da314d277121d6534b3a980b29035dcd51e6744bd79075a6ce8fa4eb8d/pytest-8.4.2-py3-none-any.whl", hash = "sha256:872f880de3fc3a5bdc88a11b39c9710c3497a547cfa9320bc3c5e62fbf272e79", size = 365750 },
 ]
 
+[[package]]
+name = "pytest-asyncio"
+version = "1.1.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "pytest" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/4e/51/f8794af39eeb870e87a8c8068642fc07bce0c854d6865d7dd0f2a9d338c2/pytest_asyncio-1.1.0.tar.gz", hash = "sha256:796aa822981e01b68c12e4827b8697108f7205020f24b5793b3c41555dab68ea", size = 46652 }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/c7/9d/bf86eddabf8c6c9cb1ea9a869d6873b46f105a5d292d3a6f7071f5b07935/pytest_asyncio-1.1.0-py3-none-any.whl", hash = "sha256:5fe2d69607b0bd75c656d1211f969cadba035030156745ee09e7d71740e58ecf", size = 15157 },
+]
+
 [[package]]
 name = "python-dateutil"
 version = "2.9.0.post0"
@@ -1502,6 +1514,7 @@ installer = [
 [package.dev-dependencies]
 dev = [
     { name = "pytest" },
+    { name = "pytest-asyncio" },
     { name = "ruff" },
 ]
 
@@ -1544,6 +1557,7 @@ provides-extras = ["installer"]
 [package.metadata.requires-dev]
 dev = [
     { name = "pytest", specifier = ">=8.4.2" },
+    { name = "pytest-asyncio", specifier = ">=1.1.0" },
     { name = "ruff", specifier = ">=0.13.0" },
 ]