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

Refactor: make classes more modular

Browse source
This commit is contained in:
arabcoders 2025-12-19 19:51:26 +03:00
parent 71cdcc220f
commit 5a5f16c9f8
13 changed files with 81 additions and 56 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
app/library/BackgroundWorker.py
View file

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

11
app/library/DataStore.py
View file

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

12
app/library/DownloadQueue.py
View file

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

2
app/library/Events.py
View file

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

7
app/library/HttpAPI.py
View file

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

7
app/library/HttpSocket.py
View file

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

8
app/library/ItemDTO.py
View file

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

17
app/library/sqlite_store.py
View file

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

19
app/main.py
View file

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

18
app/routes/api/history.py
View file

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

8
app/routes/socket/connection.py
View file

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

14
app/tests/test_datastore_pagination.py
View file

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

12
ui/app/stores/SocketStore.ts
View file

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