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

refactor: clean up sqlalchemy connection maker

Browse source
This commit is contained in:
arabcoders 2026-01-18 19:18:00 +03:00
parent a85ca93d5a
commit f98acbaa42
6 changed files with 54 additions and 35 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

44
app/library/Events.py
View file

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

3
app/library/Utils.py
View file

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

18
app/library/sqlite_store.py
View file

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

18
app/tests/test_events.py
View file

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

4
app/tests/test_update_checker.py
View file

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

2
pyproject.toml
View file

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