diff --git a/app/library/DataStore.py b/app/library/DataStore.py index 00642512..08872999 100644 --- a/app/library/DataStore.py +++ b/app/library/DataStore.py @@ -1,4 +1,3 @@ -import asyncio import copy import logging from collections import OrderedDict @@ -49,17 +48,11 @@ def _strip_transient_fields(item: ItemDTO) -> ItemDTO: class DataStore: - """ - In-memory queue cache + optional write-behind to persistence. - History reads go straight to persistence. - """ - def __init__(self, type: StoreType, connection: SqliteStore): self._type = type self._connection: SqliteStore = connection self._dict: OrderedDict[str, Download] = OrderedDict() - # cache helpers async def load(self) -> None: saved = await self._connection.fetch_saved(str(self._type)) for key, item in saved: @@ -82,21 +75,27 @@ class DataStore: if not key and not url: msg = "key or url must be provided." raise KeyError(msg) + for i in self._dict: if (key and self._dict[i].info._id == key) or (url and self._dict[i].info.url == url): return self._dict[i] + msg: str = f"{key=} or {url=} not found." raise KeyError(msg) def get_item(self, **kwargs) -> Download | None: if not kwargs: return None + for i in self._dict: if not self._dict[i].info: continue + info = self._dict[i].info.__dict__ + if any(matches_condition(key, value, info) for key, value in kwargs.items()): return self._dict[i] + return None async def get_by_id(self, id: str) -> Download | None: @@ -107,7 +106,7 @@ class DataStore: if item := await self._connection.get_by_id(str(self._type), id): download = Download(info=item) self._dict[id] = download - return download + return self._dict[id] return None @@ -117,18 +116,18 @@ class DataStore: async def put(self, value: Download, no_notify: bool = False) -> Download: _ = no_notify self._dict[value.info._id] = value - asyncio.create_task(self._connection.enqueue_upsert(str(self._type), _strip_transient_fields(value.info))) - return value + await self._connection.enqueue_upsert(str(self._type), _strip_transient_fields(value.info)) + return self._dict[value.info._id] - def delete(self, key: str) -> None: + async def delete(self, key: str) -> None: self._dict.pop(key, None) - asyncio.create_task(self._connection.enqueue_delete(str(self._type), key)) + await self._connection.enqueue_delete(str(self._type), key) def next(self): return next(iter(self._dict.items())) def empty(self) -> bool: - return not bool(self._dict) + return 0 == len(self._dict) def has_downloads(self) -> bool: return any(self._dict[key].info.auto_start and self._dict[key].started() is False for key in self._dict) diff --git a/app/library/DownloadQueue.py b/app/library/DownloadQueue.py index 0e6f2716..e34eea5c 100644 --- a/app/library/DownloadQueue.py +++ b/app/library/DownloadQueue.py @@ -165,8 +165,6 @@ class DownloadQueue(metaclass=Singleton): """ Initialize the download queue. """ - await self.done.test() - await self.done.load() await self.queue.load() LOG.info( f"Using '{self.config.max_workers}' workers for downloading and '{self.config.max_workers_per_extractor}' per extractor." @@ -904,7 +902,7 @@ class DownloadQueue(metaclass=Singleton): else: await item.close() LOG.debug(f"Deleting from queue {item_ref}") - self.queue.delete(id) + await self.queue.delete(id) self._notify.emit( Events.ITEM_CANCELLED, data=item.info, @@ -986,7 +984,7 @@ class DownloadQueue(metaclass=Singleton): except Exception as e: LOG.error(f"Unable to remove '{itemRef}' local file '{filename}'. {e!s}") - self.done.delete(id) + await self.done.delete(id) _status: str = "Removed" if removed_files > 0 else "Cleared" self._notify.emit( @@ -1169,7 +1167,7 @@ class DownloadQueue(metaclass=Singleton): if self.queue.exists(key=id): LOG.debug(f"Download Task '{id}' is completed. Removing from queue.") - self.queue.delete(key=id) + await self.queue.delete(key=id) nTitle: str | None = None nMessage: str | None = None @@ -1331,7 +1329,7 @@ class DownloadQueue(metaclass=Singleton): message=f"'{item_name}' record removed from history.", ) titles.append(item_name) - self.done.delete(key) + await self.done.delete(key) if titles: LOG.info(f"Automatically cleared '{', '.join(titles)}' from download history due to age.") diff --git a/app/library/config.py b/app/library/config.py index a4a98b2e..bf1a82bf 100644 --- a/app/library/config.py +++ b/app/library/config.py @@ -311,20 +311,20 @@ class Config(metaclass=Singleton): def __init__(self, is_native: bool = False): baseDefaultPath: str = str(Path(__file__).parent.parent.parent.absolute()) - self.is_native = is_native - self.temp_path = os.environ.get("YTP_TEMP_PATH", None) or str(Path(baseDefaultPath) / "var" / "tmp") self.config_path = os.environ.get("YTP_CONFIG_PATH", None) or str(Path(baseDefaultPath) / "var" / "config") - self.download_path = os.environ.get("YTP_DOWNLOAD_PATH", None) or str( - Path(baseDefaultPath) / "var" / "downloads" - ) - self.app_path = Path(__file__).parent.parent.absolute() - envFile: str = Path(self.config_path) / ".env" if envFile.exists(): logging.info(f"Loading environment variables from '{envFile}'.") load_dotenv(envFile) + self.is_native = is_native + self.temp_path = os.environ.get("YTP_TEMP_PATH", None) or str(Path(baseDefaultPath) / "var" / "tmp") + self.download_path = os.environ.get("YTP_DOWNLOAD_PATH", None) or str( + Path(baseDefaultPath) / "var" / "downloads" + ) + self.app_path = Path(__file__).parent.parent.absolute() + for k, v in self._get_attributes().items(): if k.startswith("_") or k in self._manual_vars: continue diff --git a/app/library/sqlite_store.py b/app/library/sqlite_store.py index 36a9e63f..536cf8cf 100644 --- a/app/library/sqlite_store.py +++ b/app/library/sqlite_store.py @@ -40,11 +40,6 @@ class _Op: class SqliteStore(metaclass=ThreadSafe): - """ - Async persistence layer with back-pressure and write-behind queue. - Singleton per process (ThreadSafe). Owns its aiosqlite connection. - """ - @staticmethod def get_instance(db_path: str | None = None) -> "SqliteStore": return SqliteStore(db_path) diff --git a/app/main.py b/app/main.py index 9ec841dc..91700480 100644 --- a/app/main.py +++ b/app/main.py @@ -134,6 +134,7 @@ class Main: def started(_): LOG.info("=" * 40) LOG.info(f"YTPTube {self._config.app_version} - started on http://{host}:{port}{self._config.base_path}") + LOG.info(f"Download path: {self._config.download_path}") if self._config.is_native: LOG.info("Running in native mode.") LOG.info("=" * 40) diff --git a/app/scripts/seed_db.py b/app/scripts/seed_db.py index 6871118c..d3539104 100644 --- a/app/scripts/seed_db.py +++ b/app/scripts/seed_db.py @@ -28,6 +28,7 @@ LOG = logging.getLogger("seed_db") USED_IDS: set[str] = set() + def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Seed history/queue with synthetic records for performance testing.") parser.add_argument( @@ -141,13 +142,21 @@ def _build_row( "live_in": None, "file_size": stat.st_size, "options": {}, - "extras": {}, + "extras": { + "source_name": "Tester", + "source_id": "0ddc40f3-9227-4ffa-9f3c-c52bcf33ef2f", + "source_handler": "YoutubeHandler", + "metadata": { + "published": formatdate(timestamp.timestamp()), + }, + "is_video": True, + "is_audio": True, + }, "cli": "", "auto_start": True, "is_archivable": True, "is_archived": True, "archive_id": archive_id, - "sidecar": {}, "tmpfilename": None, "filename": filename, "total_bytes": stat.st_size, @@ -157,8 +166,6 @@ def _build_row( "percent": 100, "speed": None, "eta": None, - "_recomputed": True, - "_archive_file": None, } created_at = timestamp.strftime("%Y-%m-%d %H:%M:%S") diff --git a/app/tests/test_datastore.py b/app/tests/test_datastore.py index b2e5be9c..a45ba191 100644 --- a/app/tests/test_datastore.py +++ b/app/tests/test_datastore.py @@ -128,7 +128,7 @@ class TestDataStore: assert row["id"] == item._id # Delete and ensure removal - store.delete(item._id) + await store.delete(item._id) await asyncio.sleep(0) await db.flush() cur2 = await db._conn.execute("SELECT * FROM history WHERE id=?", (item._id,)) @@ -638,7 +638,7 @@ class TestDataStore: store = DataStore(StoreType.QUEUE, db) # Should not raise error - store.delete("nonexistent_id") + await store.delete("nonexistent_id") # Verify nothing was deleted from database await store._connection.flush()