refactor: settings reads the image monitor through its module

get_settings located the ImageMonitor singleton by scanning sys.modules
for any module with 'image_monitor' in its name, a workaround for the
package being importable under two identities. It now uses the same
try-package / except-top-level import as the rest of the API
(_current_image_monitor) and falls back to the configured roots when no
monitor is running.
This commit is contained in:
Vito Sansevero
2026-10-08 18:01:34 -07:00
parent 38f4c0af43
commit 879247f1ef
2 changed files with 52 additions and 24 deletions
+10 -24
View File
@@ -354,29 +354,6 @@ class AdminRoutesMixin:
try:
from ..config import PromptManagerConfig, GalleryConfig
# Get monitored directories from image monitor singleton if available
monitored_dirs = []
try:
import sys
monitor_module = None
for mod_name in list(sys.modules.keys()):
if "image_monitor" in mod_name and hasattr(
sys.modules[mod_name], "_monitor_instance"
):
monitor_module = sys.modules[mod_name]
break
if monitor_module and monitor_module._monitor_instance is not None:
monitored_dirs = getattr(
monitor_module._monitor_instance, "monitored_directories", []
)
elif GalleryConfig.MONITORING_DIRECTORIES:
monitored_dirs = GalleryConfig.MONITORING_DIRECTORIES
except Exception:
if GalleryConfig.MONITORING_DIRECTORIES:
monitored_dirs = GalleryConfig.MONITORING_DIRECTORIES
root_paths = [
self._public_path(d) for d in GalleryConfig.MONITORING_DIRECTORIES
]
@@ -389,7 +366,7 @@ class AdminRoutesMixin:
"gallery_root_paths": root_paths,
"gallery_root_path": root_paths[0] if root_paths else "",
"monitored_directories": [
self._public_path(d) for d in monitored_dirs
self._public_path(d) for d in self._monitored_directories()
],
},
}
@@ -401,6 +378,15 @@ class AdminRoutesMixin:
status=500,
)
def _monitored_directories(self):
"""Directories the running image monitor watches, else the configured roots."""
from ..config import GalleryConfig
monitor = _current_image_monitor()
if monitor is not None:
return list(getattr(monitor, "monitored_directories", []))
return list(GalleryConfig.MONITORING_DIRECTORIES)
async def save_settings(self, request):
"""Save settings."""
try:
+42
View File
@@ -562,5 +562,47 @@ class TestDeleteDuplicatesSyncBody(AdminAPITestCase):
self.assertEqual(result["deleted_count"], 0)
class TestSettingsReadMonitorThroughModule(AdminAPITestCase):
"""get_settings reaches the image monitor through utils.image_monitor."""
def test_no_sys_modules_scan(self):
import inspect
from py.api.admin import AdminRoutesMixin
source = inspect.getsource(AdminRoutesMixin.get_settings)
self.assertNotIn("sys.modules", source)
async def test_reports_directories_of_running_monitor(self):
import utils.image_monitor as im_mod
from utils.image_monitor import get_image_monitor
orig = im_mod._monitor_instance
im_mod._monitor_instance = None
self.addCleanup(setattr, im_mod, "_monitor_instance", orig)
monitor = get_image_monitor(MagicMock(), MagicMock())
monitor.monitored_directories = [str(self.output_dir)]
resp = await self.client.request("GET", "/prompt_manager/settings")
data = await resp.json()
self.assertEqual(resp.status, 200)
self.assertEqual(data["settings"]["monitored_directories"], ["output"])
async def test_falls_back_to_configured_roots_without_monitor(self):
import utils.image_monitor as im_mod
orig = im_mod._monitor_instance
im_mod._monitor_instance = None
self.addCleanup(setattr, im_mod, "_monitor_instance", orig)
GalleryConfig.MONITORING_DIRECTORIES = [str(self.output_dir)]
resp = await self.client.request("GET", "/prompt_manager/settings")
data = await resp.json()
self.assertEqual(data["settings"]["monitored_directories"], ["output"])
self.assertEqual(data["settings"]["gallery_root_paths"], ["output"])
if __name__ == "__main__":
unittest.main()