Refactor version counts into stats repository

Author: rsousa88

app/main.py

@@ -40,6 +40,7 @@
     ScheduleRepository,
     ScheduleViewRepository,
     SettingsRepository,
+    StatsRepository,
     TagRepository,
     UpdateStatusRepository,
     WebuiUrlRepository,
@@ -543,6 +544,7 @@ def decorated_function(*args, **kwargs):
 schedule_view_repo = ScheduleViewRepository(get_db)
 host_metrics_repo = HostMetricsRepository(get_db)
 login_repo = LoginRepository(get_db)
+stats_repo = StatsRepository(get_db)
 
 # Container update management functions
 def check_for_update(container, client) -> Tuple[bool, Optional[str], Optional[str], Optional[str]]:
@@ -3768,14 +3770,8 @@ def get_version():
 
     # Count schedules and hosts
     try:
-        conn = get_db()
-        cursor = conn.cursor()
-        cursor.execute('SELECT COUNT(*) FROM schedules WHERE enabled = 1')
-        active_schedules = cursor.fetchone()[0]
-        cursor.execute('SELECT COUNT(*) FROM hosts WHERE enabled = 1')
-        active_hosts = cursor.fetchone()[0]
-        conn.close()
-    except:
+        active_schedules, active_hosts = stats_repo.get_active_counts()
+    except Exception:
         active_schedules = 0
         active_hosts = 0
 

app/repositories/__init__.py

@@ -9,6 +9,7 @@
 from .schedules import ScheduleRepository
 from .schedules_view import ScheduleViewRepository
 from .settings import SettingsRepository
+from .stats import StatsRepository
 from .tags import TagRepository
 from .update_status import UpdateStatusRepository
 from .users import UserRepository
@@ -26,6 +27,7 @@
     'ScheduleRepository',
     'ScheduleViewRepository',
     'SettingsRepository',
+    'StatsRepository',
     'TagRepository',
     'UpdateStatusRepository',
     'UserRepository',

app/repositories/stats.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+import sqlite3
+from typing import Callable, Tuple
+
+
+class StatsRepository:
+    """Repository for application statistics queries."""
+
+    def __init__(self, db_factory: Callable[[], sqlite3.Connection]):
+        self._db_factory = db_factory
+
+    def get_active_counts(self) -> Tuple[int, int]:
+        conn = self._db_factory()
+        try:
+            cursor = conn.cursor()
+            cursor.execute('SELECT COUNT(*) FROM schedules WHERE enabled = 1')
+            active_schedules = cursor.fetchone()[0]
+            cursor.execute('SELECT COUNT(*) FROM hosts WHERE enabled = 1')
+            active_hosts = cursor.fetchone()[0]
+            return active_schedules, active_hosts
+        finally:
+            conn.close()