From f98acbaa428dfff755594146adbb0dc654803f6c Mon Sep 17 00:00:00 2001 From: arabcoders Date: Sun, 18 Jan 2026 19:18:00 +0300 Subject: [PATCH] refactor: clean up sqlalchemy connection maker --- app/library/Events.py | 44 ++++++++++++++++++-------------- app/library/Utils.py | 3 +++ app/library/sqlite_store.py | 18 +++++++++---- app/tests/test_events.py | 18 +++++++------ app/tests/test_update_checker.py | 4 +-- pyproject.toml | 2 +- 6 files changed, 54 insertions(+), 35 deletions(-) diff --git a/app/library/Events.py b/app/library/Events.py index b0f57aa0..fae7f403 100644 --- a/app/library/Events.py +++ b/app/library/Events.py @@ -219,7 +219,7 @@ class EventBus(metaclass=Singleton): """ def __init__(self): - self._listeners: dict[str, list[str, EventListener]] = {} + self._listeners: dict[str, list[tuple[str, EventListener]]] = {} "The listeners for the events." self.debug: bool = False @@ -275,9 +275,10 @@ class EventBus(metaclass=Singleton): continue if e not in self._listeners: - self._listeners[e] = {} + self._listeners[e] = [] - self._listeners[e][name] = EventListener(name, callback) + self._listeners[e] = [(n, listener) for n, listener in self._listeners[e] if n != name] + self._listeners[e].append((name, EventListener(name, callback))) LOG.debug(f"'{name}' subscribed to '{event}'.") @@ -300,9 +301,11 @@ class EventBus(metaclass=Singleton): events = [] for e in event: - if e in self._listeners and name in self._listeners[e]: - events.append(e) - del self._listeners[e][name] + if e in self._listeners: + original_len: int = len(self._listeners[e]) + self._listeners[e] = [(n, listener) for n, listener in self._listeners[e] if n != name] + if len(self._listeners[e]) < original_len: + events.append(e) if len(events) > 0: LOG.debug(f"'{name}' unsubscribed from '{events}'.") @@ -337,22 +340,25 @@ class EventBus(metaclass=Singleton): try: loop = asyncio.get_running_loop() - for handler in self._listeners[event].values(): - try: - if handler.is_coroutine: - coro = handler.call_back(ev, handler.name, **kwargs) - if asyncio.iscoroutine(coro): - loop.create_task(coro) + async def execute_handlers(): + for _, handler in self._listeners[event]: + try: + if handler.is_coroutine: + coro = handler.call_back(ev, handler.name, **kwargs) + if asyncio.iscoroutine(coro): + await coro + else: + LOG.warning(f"Expected coroutine from async handler '{handler.name}', got {type(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}'.") + await self._call(handler, 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}'.") + + loop.create_task(execute_handlers()) except RuntimeError: LOG.debug(f"No event loop detected - using BackgroundWorker for {len(self._listeners[event])} handlers") - for handler in self._listeners[event].values(): + for _, handler in self._listeners[event]: try: if not self._offload: self._offload = BackgroundWorker.get_instance() diff --git a/app/library/Utils.py b/app/library/Utils.py index 00a50b03..b62ce94f 100644 --- a/app/library/Utils.py +++ b/app/library/Utils.py @@ -1647,6 +1647,9 @@ def load_modules(root_path: Path, directory: Path): import pkgutil package_name: str = str(directory.relative_to(root_path).as_posix()).replace("/", ".") + # Ensure package name starts with 'app.' for proper module resolution + if not package_name.startswith("app."): + package_name = f"app.{package_name}" for _, name, _ in pkgutil.iter_modules([directory]): full_name: str = f"{package_name}.{name}" diff --git a/app/library/sqlite_store.py b/app/library/sqlite_store.py index 3771042e..f56f09c3 100644 --- a/app/library/sqlite_store.py +++ b/app/library/sqlite_store.py @@ -13,6 +13,7 @@ from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine from sqlalchemy.ext.asyncio.engine import AsyncConnection +from .Events import EventBus, Events from .ItemDTO import ItemDTO from .operations import Operation, matches_condition from .Services import Services @@ -50,6 +51,12 @@ class SqliteStore(metaclass=ThreadSafe): def attach(self, app: web.Application): Services.get_instance().add("sqlite_store", self) + + async def handle_event(_, __): + await self.get_connection() + + EventBus.get_instance().subscribe(Events.STARTED, handle_event, "SqliteStore.get_connection") + app.on_shutdown.append(self.on_shutdown) async def on_shutdown(self, _: web.Application): @@ -60,13 +67,13 @@ class SqliteStore(metaclass=ThreadSafe): def __init__(self, db_path: str, *, max_pending: int = 200, flush_interval: float = 0.05): self._db_path: str = db_path self._engine: AsyncEngine | None = None - self._conn = None + self._conn: AsyncConnection | None = None self._sessionmaker: async_sessionmaker[AsyncSession] | None = None self._queue: asyncio.Queue[_Op] | None = None self._task: asyncio.Task | None = None + self._lock: asyncio.Lock = asyncio.Lock() self._flush_interval: float = flush_interval self._max_pending: int = max_pending - self._lock = asyncio.Lock() async def __aenter__(self) -> "SqliteStore": await self.get_connection() @@ -90,7 +97,7 @@ class SqliteStore(metaclass=ThreadSafe): """ if not self._sessionmaker: - msg = "Database connection not initialized. Call _ensure_conn() first or use within async context." + msg = "Database connection not initialized. Call get_connection() first or use within async context." raise RuntimeError(msg) return self._sessionmaker @@ -414,7 +421,8 @@ class SqliteStore(metaclass=ThreadSafe): if self._engine: await self._engine.dispose() self._engine = None - self._sessionmaker = None + + self._sessionmaker = None async def _enqueue(self, op: _Op) -> None: self._ensure_worker() @@ -555,7 +563,7 @@ class SqliteStore(metaclass=ThreadSafe): echo=False, connect_args={"check_same_thread": False, "uri": self._db_path.startswith(":memory")}, ) - self._conn: AsyncConnection = await self._engine.connect() + self._conn = await self._engine.connect() if version := await migrate.get_version(self._conn): LOG.debug(f"DB Version: '{version}'.") diff --git a/app/tests/test_events.py b/app/tests/test_events.py index 2b3d2912..53434952 100644 --- a/app/tests/test_events.py +++ b/app/tests/test_events.py @@ -683,8 +683,10 @@ class TestEventBus: assert result is bus # Should return self for chaining assert Events.TEST in bus._listeners - assert "test_subscriber" in bus._listeners[Events.TEST] - assert isinstance(bus._listeners[Events.TEST]["test_subscriber"], EventListener) + assert any(name == "test_subscriber" for name, _ in bus._listeners[Events.TEST]) + listener = next((listener for name, listener in bus._listeners[Events.TEST] if name == "test_subscriber"), None) + assert listener is not None + assert isinstance(listener, EventListener) @patch("app.library.config.Config") @patch("app.library.BackgroundWorker.BackgroundWorker") @@ -703,7 +705,7 @@ class TestEventBus: for event in events: assert event in bus._listeners - assert "multi_subscriber" in bus._listeners[event] + assert any(name == "multi_subscriber" for name, _ in bus._listeners[event]) @patch("app.library.config.Config") @patch("app.library.BackgroundWorker.BackgroundWorker") @@ -722,7 +724,7 @@ class TestEventBus: all_events = Events.get_all() for event in all_events: assert event in bus._listeners - assert "wildcard_subscriber" in bus._listeners[event] + assert any(name == "wildcard_subscriber" for name, _ in bus._listeners[event]) @patch("app.library.config.Config") @patch("app.library.BackgroundWorker.BackgroundWorker") @@ -741,7 +743,7 @@ class TestEventBus: frontend_events = Events.frontend() for event in frontend_events: assert event in bus._listeners - assert "frontend_subscriber" in bus._listeners[event] + assert any(name == "frontend_subscriber" for name, _ in bus._listeners[event]) @patch("app.library.config.Config") @patch("app.library.BackgroundWorker.BackgroundWorker") @@ -779,7 +781,7 @@ class TestEventBus: assert len(bus._listeners[Events.TEST]) == 1 # Name should be a UUID - subscriber_name = next(iter(bus._listeners[Events.TEST].keys())) + subscriber_name = bus._listeners[Events.TEST][0][0] assert len(subscriber_name) == 36 # UUID string length @patch("app.library.config.Config") @@ -796,13 +798,13 @@ class TestEventBus: # First subscribe bus.subscribe(Events.TEST, test_callback, "test_subscriber") - assert "test_subscriber" in bus._listeners[Events.TEST] + assert any(name == "test_subscriber" for name, _ in bus._listeners[Events.TEST]) # Then unsubscribe result = bus.unsubscribe(Events.TEST, "test_subscriber") assert result is bus - assert "test_subscriber" not in bus._listeners[Events.TEST] + assert not any(name == "test_subscriber" for name, _ in bus._listeners[Events.TEST]) @patch("app.library.config.Config") @patch("app.library.BackgroundWorker.BackgroundWorker") diff --git a/app/tests/test_update_checker.py b/app/tests/test_update_checker.py index 18cebf34..7a820a22 100644 --- a/app/tests/test_update_checker.py +++ b/app/tests/test_update_checker.py @@ -351,9 +351,9 @@ class TestUpdateChecker: checker.attach(app_mock) # Verify subscription was created - subscriptions = notify._listeners.get("started", {}) + subscriptions = notify._listeners.get("started", []) assert len(subscriptions) > 0, "Should have subscribed to STARTED event" - assert any("UpdateChecker.attach" in name for name in subscriptions.keys()), ( + assert any("UpdateChecker.attach" in name for name, _ in subscriptions), ( "Should have UpdateChecker.attach subscription" ) finally: diff --git a/pyproject.toml b/pyproject.toml index a419e71d..72d2b805 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -183,7 +183,7 @@ link-mode = "copy" [tool.pytest.ini_options] pythonpath = ["."] -testpaths = ["app/tests"] +testpaths = ["app/tests", "app/features"] addopts = "-v --tb=short" filterwarnings = [ "ignore:Parsing dates involving a day of month without a year:DeprecationWarning",