From 5a5f16c9f802107112307bfb0188b5a7304701e1 Mon Sep 17 00:00:00 2001 From: arabcoders Date: Fri, 19 Dec 2025 19:51:26 +0300 Subject: [PATCH] Refactor: make classes more modular --- app/library/BackgroundWorker.py | 2 ++ app/library/DataStore.py | 11 +++++++++-- app/library/DownloadQueue.py | 12 ++++++------ app/library/Events.py | 2 ++ app/library/HttpAPI.py | 7 ++----- app/library/HttpSocket.py | 7 +------ app/library/ItemDTO.py | 8 ++------ app/library/sqlite_store.py | 17 ++++++++--------- app/main.py | 19 ++++++++----------- app/routes/api/history.py | 18 +++++++++++++++++- app/routes/socket/connection.py | 8 ++++++++ app/tests/test_datastore_pagination.py | 14 ++++++++------ ui/app/stores/SocketStore.ts | 12 ++++++++---- 13 files changed, 81 insertions(+), 56 deletions(-) diff --git a/app/library/BackgroundWorker.py b/app/library/BackgroundWorker.py index 48add1d3..f5f4dea9 100644 --- a/app/library/BackgroundWorker.py +++ b/app/library/BackgroundWorker.py @@ -6,6 +6,7 @@ from queue import Empty, Queue from aiohttp import web +from .Services import Services from .Singleton import Singleton LOG: logging.Logger = logging.getLogger("BackgroundWorker") @@ -35,6 +36,7 @@ class BackgroundWorker(metaclass=Singleton): return BackgroundWorker() def attach(self, app: web.Application): + Services.get_instance().add("background_worker", self) app.on_shutdown.append(self.on_shutdown) LOG.debug("Starting background worker...") diff --git a/app/library/DataStore.py b/app/library/DataStore.py index 16491e96..33cb44d5 100644 --- a/app/library/DataStore.py +++ b/app/library/DataStore.py @@ -154,18 +154,25 @@ class DataStore: async def get_items_paginated( self, page: int = 1, per_page: int = 50, order: str = "DESC", status_filter: str | None = None - ): + ) -> tuple[list[tuple[str, Download]], int, int, int]: if page < 1: msg = "page must be >= 1" raise ValueError(msg) + if per_page < 1: msg = "per_page must be >= 1" raise ValueError(msg) + order = order.upper() if order not in ("ASC", "DESC"): msg = f"order must be 'ASC' or 'DESC', got '{order}'" raise ValueError(msg) - return await self._connection.paginate(str(self._type), page, per_page, order, status_filter) + + items, total_items, current_page, total_pages = await self._connection.paginate( + str(self._type), page, per_page, order, status_filter + ) + + return [(item_id, Download(info=item)) for item_id, item in items], total_items, current_page, total_pages async def bulk_delete(self, ids: Iterable[str]) -> int: deleted = await self._connection.bulk_delete(str(self._type), ids) diff --git a/app/library/DownloadQueue.py b/app/library/DownloadQueue.py index 86f14acf..5a1d4fbd 100644 --- a/app/library/DownloadQueue.py +++ b/app/library/DownloadQueue.py @@ -23,6 +23,7 @@ from .Events import EventBus, Events from .ItemDTO import Item, ItemDTO from .Presets import Presets from .Scheduler import Scheduler +from .Services import Services from .Singleton import Singleton from .sqlite_store import SqliteStore from .Utils import ( @@ -54,9 +55,9 @@ class DownloadQueue(metaclass=Singleton): "Configuration instance." self._notify: EventBus = EventBus.get_instance() "Event bus instance." - self.done = DataStore(type=StoreType.HISTORY, connection=SqliteStore.get_instance(config.db_file)) + self.done = DataStore(type=StoreType.HISTORY, connection=SqliteStore.get_instance()) "DataStore for the completed downloads." - self.queue = DataStore(type=StoreType.QUEUE, connection=SqliteStore.get_instance(config.db_file)) + self.queue = DataStore(type=StoreType.QUEUE, connection=SqliteStore.get_instance()) "DataStore for the download queue." self.workers = asyncio.Semaphore(self.config.max_workers) "Semaphore to limit the number of concurrent downloads." @@ -74,7 +75,7 @@ class DownloadQueue(metaclass=Singleton): self.paused.set() @staticmethod - def get_instance() -> "DownloadQueue": + def get_instance(config: Config | None = None) -> "DownloadQueue": """ Get the instance of the DownloadQueue. @@ -82,7 +83,7 @@ class DownloadQueue(metaclass=Singleton): DownloadQueue: The instance of the DownloadQueue """ - return DownloadQueue() + return DownloadQueue(config=config) def _get_limit(self, extractor: str) -> asyncio.Semaphore: """ @@ -121,6 +122,7 @@ class DownloadQueue(metaclass=Singleton): _ (web.Application): The application to attach the download queue to. """ + Services.get_instance().add("queue", self) async def event_handler(_, __): await self.initialize() @@ -1021,7 +1023,6 @@ class DownloadQueue(metaclass=Singleton): if mode in ("all", "done"): for k, v in await self.done.saved_items(): - v.get_file_sidecar() items["done"][k] = v if mode in ("all", "queue"): @@ -1034,7 +1035,6 @@ class DownloadQueue(metaclass=Singleton): if k in items["done"]: continue - v.info.get_file_sidecar() items["done"][k] = v.info return items diff --git a/app/library/Events.py b/app/library/Events.py index f7528927..39a0232a 100644 --- a/app/library/Events.py +++ b/app/library/Events.py @@ -25,6 +25,7 @@ class Events: CONNECTED: str = "connected" CONFIGURATION: str = "configuration" + ACTIVE_QUEUE: str = "active_queue" LOG_INFO: str = "log_info" LOG_WARNING: str = "log_warning" @@ -94,6 +95,7 @@ class Events: return [ Events.CONFIGURATION, Events.CONNECTED, + Events.ACTIVE_QUEUE, Events.LOG_INFO, Events.LOG_WARNING, Events.LOG_ERROR, diff --git a/app/library/HttpAPI.py b/app/library/HttpAPI.py index 287d4684..7c00ee2d 100644 --- a/app/library/HttpAPI.py +++ b/app/library/HttpAPI.py @@ -14,7 +14,6 @@ from app.library.Services import Services from .cache import Cache from .config import Config -from .DownloadQueue import DownloadQueue from .encoder import Encoder from .Events import EventBus from .router import RouteType, get_routes @@ -24,8 +23,7 @@ LOG: logging.Logger = logging.getLogger("http_api") class HttpAPI: - def __init__(self, root_path: Path, queue: DownloadQueue): - self.queue: DownloadQueue = queue or DownloadQueue.get_instance() + def __init__(self, root_path: Path): self.encoder: Encoder = Encoder() self.config: Config = Config.get_instance() self._notify: EventBus = EventBus.get_instance() @@ -33,12 +31,11 @@ class HttpAPI: self.cache = Cache() self.app: web.Application | None = None - services = Services.get_instance() + services: Services = Services.get_instance() services.add_all( { k: v for k, v in { - "queue": self.queue, "encoder": self.encoder, "config": self.config, "notify": self._notify, diff --git a/app/library/HttpSocket.py b/app/library/HttpSocket.py index 16f3e4b2..b73ac236 100644 --- a/app/library/HttpSocket.py +++ b/app/library/HttpSocket.py @@ -11,7 +11,6 @@ from app.library.Services import Services from app.library.Utils import load_modules from .config import Config -from .DownloadQueue import DownloadQueue from .encoder import Encoder from .Events import Event, EventBus, Events from .ItemDTO import Item @@ -26,19 +25,16 @@ class HttpSocket: config: Config sio: socketio.AsyncServer - queue: DownloadQueue di_context: dict[str, Any] = {} def __init__( self, root_path: Path, - queue: DownloadQueue | None = None, encoder: Encoder | None = None, config: Config | None = None, sio: socketio.AsyncServer | None = None, ): self.config = config or Config.get_instance() - self.queue = queue or DownloadQueue.get_instance() self._notify = EventBus.get_instance() self.sio = sio or socketio.AsyncServer( @@ -63,7 +59,6 @@ class HttpSocket: k: v for k, v in { "config": self.config, - "queue": self.queue, "sio": self.sio, "encoder": encoder, "notify": self._notify, @@ -104,7 +99,7 @@ class HttpSocket: async def event_handler(data: Event, _): if data and data.data: - await self.queue.add(item=Item.format(data.data)) + await Services.get_instance().get("queue").add(item=Item.format(data.data)) self._notify.subscribe(Events.ADD_URL, event_handler, f"{__class__.__name__}.add") diff --git a/app/library/ItemDTO.py b/app/library/ItemDTO.py index 62cc560b..cb75e43f 100644 --- a/app/library/ItemDTO.py +++ b/app/library/ItemDTO.py @@ -397,11 +397,8 @@ class ItemDTO: dict: The serialized item. """ - if "finished" == self.status: - if not self._recomputed: - self.archive_status() - - self.get_file_sidecar() + if "finished" == self.status and not self._recomputed: + self.archive_status() item, _ = clean_item(self.__dict__.copy(), ItemDTO.removed_fields()) return item @@ -661,4 +658,3 @@ class ItemDTO: self.get_archive_id() self.get_archive_file() self.archive_status() - self.get_file_sidecar() diff --git a/app/library/sqlite_store.py b/app/library/sqlite_store.py index 24eb933c..7a1b7df3 100644 --- a/app/library/sqlite_store.py +++ b/app/library/sqlite_store.py @@ -13,6 +13,7 @@ from aiohttp import web from .ItemDTO import ItemDTO from .operations import Operation, matches_condition +from .Services import Services from .Singleton import ThreadSafe from .Utils import init_class @@ -43,25 +44,24 @@ class _Op: class SqliteStore(metaclass=ThreadSafe): @staticmethod def get_instance(db_path: str | None = None) -> "SqliteStore": - return SqliteStore(db_path) + return SqliteStore(db_path=db_path) def attach(self, app: web.Application): - """Get/create singleton bound to db_path.""" + Services.get_instance().add("sqlite_store", self) app.on_shutdown.append(self.on_shutdown) async def on_shutdown(self, _: web.Application): - """Close singleton on app shutdown.""" LOG.debug("Shutting down SqliteStore...") await self.close() LOG.debug("SqliteStore shut down complete.") - def __init__(self, db_path: str | None = None, *, max_pending: int = 200, flush_interval: float = 0.05): - self._db_path = db_path + def __init__(self, db_path: str, *, max_pending: int = 200, flush_interval: float = 0.05): + self._db_path: str = db_path self._conn: aiosqlite.Connection | None = None self._queue: asyncio.Queue[_Op] | None = None self._task: asyncio.Task | None = None - self._flush_interval = flush_interval - self._max_pending = max_pending + self._flush_interval: float = flush_interval + self._max_pending: int = max_pending self._lock = asyncio.Lock() async def __aenter__(self) -> "SqliteStore": @@ -303,7 +303,6 @@ class SqliteStore(metaclass=ThreadSafe): return items, total_items, page, total_pages - # direct CRUD helpers (used by DataStore) async def upsert(self, type_value: str, item: ItemDTO) -> None: await self._ensure_conn() await self._upsert_now(type_value, item) @@ -441,7 +440,7 @@ class SqliteStore(metaclass=ThreadSafe): return if not self._db_path: - msg = "SqliteStore requires db_path or injected connection." + msg = "No database path specified for SqliteStore." raise RuntimeError(msg) from app.library import migrate diff --git a/app/main.py b/app/main.py index 91700480..822e9cbe 100644 --- a/app/main.py +++ b/app/main.py @@ -17,7 +17,6 @@ from aiohttp import web from app.library.BackgroundWorker import BackgroundWorker from app.library.conditions import Conditions from app.library.config import Config -from app.library.DataStore import SqliteStore from app.library.dl_fields import DLFields from app.library.DownloadQueue import DownloadQueue from app.library.Events import EventBus, Events @@ -27,6 +26,7 @@ from app.library.Notifications import Notification from app.library.Presets import Presets from app.library.Scheduler import Scheduler from app.library.Services import Services +from app.library.sqlite_store import SqliteStore from app.library.TaskDefinitions import TaskDefinitions from app.library.Tasks import Tasks @@ -42,10 +42,8 @@ class Main: self._config.set_app_path(str(ROOT_PATH)) self._app = web.Application() self._app.on_shutdown.append(self.on_shutdown) - self._background_worker = BackgroundWorker() Services.get_instance().add("app", self._app) - Services.get_instance().add("background_worker", self._background_worker) self._check_folders() @@ -56,13 +54,12 @@ class Main: except Exception: pass - self._queue = DownloadQueue(config=self._config) - self._http = HttpAPI(root_path=ROOT_PATH, queue=self._queue) - self._socket = HttpSocket(root_path=ROOT_PATH, queue=self._queue) + self._http = HttpAPI(root_path=ROOT_PATH) + self._socket = HttpSocket(root_path=ROOT_PATH) def _check_folders(self): """Check if the required folders exist and create them if they do not.""" - folders = (self._config.download_path, self._config.temp_path, self._config.config_path) + folders: tuple[str, str, str] = (self._config.download_path, self._config.temp_path, self._config.config_path) for folder in folders: folder = Path(folder) @@ -109,20 +106,20 @@ class Main: if self._config.debug: EventBus.get_instance().debug_enable() + SqliteStore.get_instance(db_path=self._config.db_file).attach(self._app) + BackgroundWorker.get_instance().attach(self._app) Scheduler.get_instance().attach(self._app) self._socket.attach(self._app) self._http.attach(self._app) - self._queue.attach(self._app) - Tasks.get_instance().attach(self._app) Presets.get_instance().attach(self._app) + Tasks.get_instance().attach(self._app) Notification.get_instance().attach(self._app) Conditions.get_instance().attach(self._app) DLFields.get_instance().attach(self._app) TaskDefinitions.get_instance().attach(self._app) - SqliteStore.get_instance().attach(self._app) - self._background_worker.attach(self._app) + DownloadQueue.get_instance().attach(self._app) EventBus.get_instance().emit( Events.LOADED, diff --git a/app/routes/api/history.py b/app/routes/api/history.py index 7229f4fc..1255f9fe 100644 --- a/app/routes/api/history.py +++ b/app/routes/api/history.py @@ -89,6 +89,16 @@ async def items_list(request: Request, queue: DownloadQueue, encoder: Encoder, c page=page, per_page=per_page, order=order, status_filter=status_filter ) + if store_type == StoreType.HISTORY: + for _, download in items: + if not download.info: + continue + + try: + download.info.sidecar = download.get_file_sidecar() + except Exception: + download.info.sidecar = {} + return web.json_response( data={ "type": store_type.value, @@ -100,7 +110,7 @@ async def items_list(request: Request, queue: DownloadQueue, encoder: Encoder, c "has_next": current_page < total_pages, "has_prev": current_page > 1, }, - "items": [item for _, item in items], + "items": [download.info for _, download in items], }, status=web.HTTPOk.status_code, dumps=encoder.encode, @@ -227,6 +237,7 @@ async def item_view(request: Request, queue: DownloadQueue, encoder: Encoder) -> info: dict = { **item.info.serialize(), "ffprobe": {}, + "sidecar": {}, } if "finished" == item.info.status and (filename := item.info.get_file()): @@ -237,6 +248,11 @@ async def item_view(request: Request, queue: DownloadQueue, encoder: Encoder) -> except Exception: pass + try: + info["sidecar"] = item.info.get_file_sidecar() + except Exception: + pass + return web.json_response(data=info, status=web.HTTPOk.status_code, dumps=encoder.encode) diff --git a/app/routes/socket/connection.py b/app/routes/socket/connection.py index d4fc85ba..38e058ae 100644 --- a/app/routes/socket/connection.py +++ b/app/routes/socket/connection.py @@ -52,6 +52,14 @@ async def connect(config: Config, queue: DownloadQueue, notify: EventBus, sid: s to=sid, ) + notify.emit( + Events.ACTIVE_QUEUE, + data={"queue": (await queue.get("queue"))["queue"]}, + title="Sending initial active queue data", + message=f"Sending active queue data to client '{sid}'.", + to=sid, + ) + @route(RouteType.SOCKET, "disconnect", "socket_disconnect") async def disconnect(sio: socketio.AsyncServer, sid: str, data: str = None): diff --git a/app/tests/test_datastore_pagination.py b/app/tests/test_datastore_pagination.py index da827824..0c8c280b 100644 --- a/app/tests/test_datastore_pagination.py +++ b/app/tests/test_datastore_pagination.py @@ -5,6 +5,7 @@ import pytest import pytest_asyncio from app.library.DataStore import DataStore, StoreType +from app.library.Download import Download from app.library.ItemDTO import ItemDTO from app.library.sqlite_store import SqliteStore @@ -75,8 +76,9 @@ class TestDataStorePagination: assert page == 1 assert total_pages == 10 for _, item in items: - assert isinstance(item, ItemDTO) - assert item._id.startswith("test-id-") + assert isinstance(item, Download) + assert isinstance(item.info, ItemDTO) + assert item.info._id.startswith("test-id-") finally: await db.close() @@ -253,7 +255,7 @@ class TestDataStorePagination: assert len(items) == 50 # First page of finished items assert total == 100 # Only 100 finished items in fixture for _item_id, item in items: - assert item.status == "finished" + assert item.info.status == "finished" # Filter for pending items only items, total, _page, _total_pages = await datastore.get_items_paginated( @@ -262,7 +264,7 @@ class TestDataStorePagination: assert len(items) == 1 assert total == 1 - assert items[0][1].status == "pending" + assert items[0][1].info.status == "pending" finally: await db.close() @@ -317,10 +319,10 @@ class TestDataStorePagination: assert total == 2 # Only 2 non-finished items assert len(items) == 2 for _item_id, item in items: - assert item.status != "finished" + assert item.info.status != "finished" # Verify we have pending and error - statuses = {item.status for _, item in items} + statuses = {item.info.status for _, item in items} assert statuses == {"pending", "error"} finally: await db.close() diff --git a/ui/app/stores/SocketStore.ts b/ui/app/stores/SocketStore.ts index bb7dfc81..bfc517c4 100644 --- a/ui/app/stores/SocketStore.ts +++ b/ui/app/stores/SocketStore.ts @@ -118,10 +118,6 @@ export const useSocketStore = defineStore('socket', () => { config.add('folders', json.data.folders) } - if (json.data?.queue) { - stateStore.addAll('queue', json.data.queue || {}) - } - if (typeof json.data?.history_count === 'number') { stateStore.setHistoryCount(json.data.history_count) } @@ -129,6 +125,14 @@ export const useSocketStore = defineStore('socket', () => { error.value = null; }) + on('active_queue', stream => { + const json = JSON.parse(stream); + if (!json?.data?.queue) { + return; + } + stateStore.addAll('queue', json.data.queue || {}) + }) + on('item_added', stream => { const json = JSON.parse(stream); stateStore.add('queue', json.data._id, json.data);