danvics/ytptube
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Refactor: do not load history items in DownloadQueue

Browse source
This commit is contained in:
arabcoders 2025-12-18 17:08:18 +03:00
parent 7a85bcdb5c
commit 50f50bed9f
7 changed files with 37 additions and 37 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

25
app/library/DataStore.py
View file

@ -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)

10
app/library/DownloadQueue.py
View file

@ -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.")

14
app/library/config.py
View file

@ -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

5
app/library/sqlite_store.py
View file

@ -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)

1
app/main.py
View file

@ -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)

15
app/scripts/seed_db.py
View file

@ -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")

4
app/tests/test_datastore.py
View file

@ -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()