refactor: extract API routes into core/routes package

Move the inline route definitions out of __init__.py into a new core/routes
package (registry, base_models, helpers, settings, version). Add the
JobProgressTracker utility in core/progress.py and extract version discovery
logic into core/version.py. Update tests to cover the refactored structure.
This commit is contained in:
Dariusz L
2026-08-01 21:57:51 +02:00
parent 422899e91b
commit b12cc65a10
10 changed files with 5459 additions and 5219 deletions
+16 -5196
View File
File diff suppressed because it is too large Load Diff
+116 -5
View File
@@ -1,9 +1,7 @@
"""
Progress Reporting Module
Unified utility for reporting progress of asynchronous search and download operations.
"""
"""Progress tracking and reporting utilities."""
import threading
import time
from typing import Any, Callable, Dict, Optional
from .log_system import create_module_logger
@@ -11,6 +9,119 @@ from .log_system import create_module_logger
log = create_module_logger(__name__)
class JobProgressTracker:
"""Helper class for thread-safe job progress tracking and cancellation management."""
def __init__(self, default_message="Processing..."):
self.lock = threading.Lock()
self.progress = {}
self.cancelled = set()
self.default_message = default_message
def cleanup(self, max_age_seconds=300):
cutoff = time.time() - max_age_seconds
with self.lock:
expired = [
pid
for pid, data in self.progress.items()
if data.get("updated_at", data.get("created_at", 0)) < cutoff
]
for pid in expired:
self.progress.pop(pid, None)
self.cancelled.difference_update(expired)
def update(self, progress_id, source=None, stage=None, message=None, percent=None, status=None, **payload):
if not progress_id:
return
if source is not None:
payload["source"] = source
if stage is not None:
payload["stage"] = stage
if message is not None:
payload["message"] = message
if percent is not None:
payload["percent"] = percent
if status is not None:
payload["status"] = status
now = time.time()
with self.lock:
current = self.progress.get(progress_id, {})
# If the job was cancelled, force it to remain cancelled.
if progress_id in self.cancelled and payload.get("status") != "cancelled":
payload["status"] = "cancelled"
payload["stage"] = "cancelled"
payload["message"] = "Cancelled"
payload["percent"] = 100
payload["cancelled"] = True
# Normalize percent if present
if "percent" in payload and payload["percent"] is not None:
try:
payload["percent"] = max(0.0, min(100.0, float(payload["percent"])))
except (TypeError, ValueError):
pass
self.progress[progress_id] = {
"created_at": now,
"message": self.default_message,
**current,
**payload,
"progress_id": progress_id,
"updated_at": now,
}
def update_from_payload(
self,
progress_id: Optional[str],
progress_payload: Dict[str, Any],
default_stage: str = "running",
) -> None:
if not progress_id or not isinstance(progress_payload, dict):
return
data = dict(progress_payload)
status = data.pop("status", None)
stage = data.pop("stage", None) or default_stage
if not status:
if stage in ("done", "completed"):
status = "completed"
elif stage == "cancelled":
status = "cancelled"
else:
status = "running"
self.update(progress_id, status=status, stage=stage, **data)
def is_cancelled(self, progress_id) -> bool:
if not progress_id:
return False
with self.lock:
return progress_id in self.cancelled
def mark_cancelled(self, progress_id, cancel_message="Cancelled") -> bool:
self.cleanup()
with self.lock:
current = self.progress.get(progress_id, {})
self.cancelled.add(progress_id)
self.progress[progress_id] = {
"created_at": time.time(),
**current,
"progress_id": progress_id,
"status": "cancelled",
"stage": "cancelled",
"message": cancel_message,
"percent": 100,
"cancelled": True,
"updated_at": time.time(),
}
return True
def get(self, progress_id):
with self.lock:
val = self.progress.get(progress_id)
return dict(val) if val else None
def report_progress(
progress_callback: Optional[Callable[[Dict[str, Any]], None]],
stage: str,
+1
View File
@@ -0,0 +1 @@
"""HTTP route registration modules for the Model Resolver backend."""
+35
View File
@@ -0,0 +1,35 @@
"""Routes for the bundled base-model catalog."""
import asyncio
from ..sources.popular import (
get_base_models_config,
get_base_models_status,
update_base_models_from_remote,
)
def register_base_model_routes(routes, web, json_api_endpoint):
"""Register base-model catalog endpoints."""
@routes.get("/model_resolver/base-models")
@json_api_endpoint("base-models")
async def get_base_models(request):
"""Return the bundled base-model configuration."""
data = get_base_models_config()
return web.json_response(data)
@routes.get("/model_resolver/base-models/status")
@json_api_endpoint("base-models status")
async def get_base_models_status_route(request):
"""Get local and optional remote base-model status."""
check_remote = request.query.get("check_remote") == "1"
status = await asyncio.to_thread(get_base_models_status, check_remote)
return web.json_response(status)
@routes.post("/model_resolver/base-models/update")
@json_api_endpoint("base-models update")
async def update_base_models_route(request):
"""Update the base-model list from CivitAI."""
status = await asyncio.to_thread(update_base_models_from_remote)
return web.json_response(status)
+143
View File
@@ -0,0 +1,143 @@
"""Shared helpers used by the HTTP route modules."""
import asyncio
import threading
from functools import wraps
def create_route_helpers(web, logger, load_settings, hash_calculation_cancelled):
"""Create route helpers bound to the current ComfyUI server context."""
def json_api_endpoint(error_prefix, return_success_on_error=False):
def decorator(func):
@wraps(func)
async def wrapper(request, *args, **kwargs):
try:
return await func(request, *args, **kwargs)
except Exception as exc:
logger.error(
f"Model Resolver {error_prefix} error: {exc}",
exc_info=True,
)
response_data = {"error": str(exc)}
if return_success_on_error:
response_data["success"] = False
return web.json_response(response_data, status=500)
return wrapper
return decorator
def get_progress_response(
tracker,
request,
param_name="progress_id",
not_found_payload=None,
not_found_status=200,
found_wrapper=None,
):
job_id = request.match_info.get(param_name, "").strip()
if not job_id:
return web.json_response(
{"error": f"{param_name} is required"},
status=400,
)
tracker.cleanup()
progress = tracker.get(job_id)
if not progress:
payload = not_found_payload or {"error": "progress not found"}
return web.json_response(payload, status=not_found_status)
if found_wrapper:
return web.json_response(found_wrapper(progress))
return web.json_response(progress)
def cancel_progress_response(
tracker,
request,
param_name="progress_id",
cancel_message="Cancelled",
):
job_id = request.match_info.get(param_name, "").strip()
if not job_id:
return web.json_response(
{"error": f"{param_name} is required"},
status=400,
)
cancelled = tracker.mark_cancelled(job_id, cancel_message)
return web.json_response(
{
"success": True,
"cancelled": cancelled,
"progress_id": job_id,
}
)
def run_in_background_thread(
tracker,
progress_id,
task_func,
on_success,
on_cancel=None,
on_error=None,
error_log_msg="Background task failed",
):
def wrapper():
try:
result = task_func()
if tracker.is_cancelled(progress_id):
if on_cancel:
on_cancel(result)
else:
tracker.update(
progress_id,
status="cancelled",
stage="cancelled",
message="Task cancelled",
)
return
on_success(result)
except (hash_calculation_cancelled, asyncio.CancelledError):
if on_cancel:
on_cancel()
else:
tracker.update(
progress_id,
status="cancelled",
stage="cancelled",
message="Task cancelled",
)
except Exception as exc:
logger.exception(f"{error_log_msg}: {exc}")
if on_error:
on_error(exc)
else:
tracker.update(
progress_id,
status="error",
stage="error",
message=str(exc) or "Task failed",
percent=100,
error=str(exc) or "Task failed",
)
threading.Thread(target=wrapper, daemon=True).start()
async def get_override_settings_from_request(request):
settings = await asyncio.to_thread(load_settings)
if request.method == "POST":
try:
payload = await request.json()
if isinstance(payload, dict) and "aria2c_path" in payload:
settings = dict(settings)
settings["aria2c_path"] = payload.get("aria2c_path", "")
except Exception:
pass
return settings
return (
json_api_endpoint,
get_progress_response,
cancel_progress_response,
run_in_background_thread,
get_override_settings_from_request,
)
File diff suppressed because it is too large Load Diff
+147
View File
@@ -0,0 +1,147 @@
"""Settings and backend-log HTTP routes."""
import asyncio
import os
import time
from ..log_system import LogLevel
from ..log_system import logger as backend_log_controller
from ..log_system.config import LOG_LEVEL as BACKEND_DEFAULT_LOG_LEVEL
from ..log_system.logger import parse_rotated_log_filename
from ..settings import bool_setting as resolver_bool_setting
from ..settings import get_settings_schema
from ..settings import load_settings as load_resolver_settings
from ..settings import save_settings as save_resolver_settings
def _flush_backend_log_handlers():
for handler in getattr(backend_log_controller, "file_handlers", {}).values():
try:
handler.flush()
except Exception:
pass
def _backend_log_sort_key(path):
name = os.path.basename(path)
parsed_name = parse_rotated_log_filename(name)
if parsed_name is None:
return (name.lower(), 999)
base_name, rotation = parsed_name
return (base_name.lower(), rotation)
def _collect_backend_log_files(log_dir):
if not log_dir or not os.path.isdir(log_dir):
return []
files = []
for entry in os.listdir(log_dir):
name = str(entry or "")
if not name.startswith("azlogs_"):
continue
if parse_rotated_log_filename(name) is None:
continue
path = os.path.abspath(os.path.join(log_dir, name))
try:
if os.path.commonpath([log_dir, path]) != log_dir:
continue
except ValueError:
continue
if os.path.isfile(path):
files.append(path)
return sorted(files, key=_backend_log_sort_key)
def _build_backend_log_export():
_flush_backend_log_handlers()
raw_log_dir = str(backend_log_controller.config.get("log_dir") or "")
log_dir = os.path.abspath(raw_log_dir) if raw_log_dir else ""
exported_at = time.strftime("%Y-%m-%d %H:%M:%S")
lines = [
"Model Resolver Backend Logs",
f"Exported: {exported_at}",
f"Log directory: {log_dir or 'not configured'}",
f"File logging: {bool(backend_log_controller.config.get('log_to_file'))}",
"",
]
log_files = _collect_backend_log_files(log_dir)
if not log_files:
lines.append("No backend log files found.")
lines.append("")
return "\n".join(lines)
for path in log_files:
name = os.path.basename(path)
try:
stat = os.stat(path)
modified_at = time.strftime(
"%Y-%m-%d %H:%M:%S",
time.localtime(stat.st_mtime),
)
lines.append(
f"===== {name} ({stat.st_size} bytes, modified {modified_at}) ====="
)
with open(path, encoding="utf-8", errors="replace") as log_file:
lines.append(log_file.read().rstrip())
except OSError as exc:
lines.append(f"===== {name} =====")
lines.append(f"Could not read log file: {exc}")
lines.append("")
return "\n".join(lines)
def _log_level_setting(value, default: str = BACKEND_DEFAULT_LOG_LEVEL) -> LogLevel:
normalized = str(value or default or "INFO").strip().upper()
if hasattr(LogLevel, normalized):
return getattr(LogLevel, normalized)
fallback = str(default or "INFO").strip().upper()
return getattr(LogLevel, fallback, LogLevel.INFO)
def _apply_backend_logging_settings(settings: dict) -> None:
enabled = resolver_bool_setting(settings.get("backend_logs_enabled"), True)
level = _log_level_setting(settings.get("backend_log_level"))
backend_log_controller.set_enabled(enabled)
backend_log_controller.set_global_level(level)
def register_settings_routes(routes, web, json_api_endpoint):
"""Register settings persistence and backend-log export endpoints."""
_apply_backend_logging_settings(load_resolver_settings())
@routes.get("/model_resolver/logs/backend/export")
@json_api_endpoint("backend log export")
async def export_backend_logs_route(request):
"""Download Model Resolver backend logs as a text file."""
content = await asyncio.to_thread(_build_backend_log_export)
filename = f"model_resolver_backend_logs_{time.strftime('%Y%m%d_%H%M%S')}.txt"
return web.Response(
text=content,
content_type="text/plain",
headers={
"Content-Disposition": f'attachment; filename="{filename}"',
"Cache-Control": "no-store",
},
)
@routes.get("/model_resolver/settings")
@json_api_endpoint("settings GET")
async def get_settings_route(request):
"""Return persisted settings (API keys, preferences)."""
data = await asyncio.to_thread(load_resolver_settings)
return web.json_response({"settings": data, "schema": get_settings_schema()})
@routes.post("/model_resolver/settings")
@json_api_endpoint("settings POST")
async def save_settings_route(request):
"""Persist settings (API keys, preferences) to disk."""
payload = await request.json()
if not isinstance(payload, dict):
return web.json_response({"error": "Expected JSON object"}, status=400)
settings = await asyncio.to_thread(save_resolver_settings, payload)
_apply_backend_logging_settings(settings)
return web.json_response({"success": True})
+17
View File
@@ -0,0 +1,17 @@
"""Version endpoint registration."""
import asyncio
from .. import version as version_service
def register_version_routes(routes, web, json_api_endpoint):
"""Register the endpoint exposing local and remote project versions."""
@routes.get("/model_resolver/version")
@json_api_endpoint("version")
async def get_project_version(request):
"""Return the installed version and the latest GitHub version."""
return web.json_response(
await asyncio.to_thread(version_service._get_project_version_info)
)
+92
View File
@@ -0,0 +1,92 @@
"""Project version discovery and update checks."""
import os
import re
import threading
import time
from typing import Any
PROJECT_VERSION_FILE = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"pyproject.toml",
)
PROJECT_GITHUB_URL = "https://github.com/Azornes/Comfyui-Model-Resolver"
PROJECT_GITHUB_PYPROJECT_URL = (
"https://raw.githubusercontent.com/Azornes/Comfyui-Model-Resolver/main/pyproject.toml"
)
PROJECT_VERSION_CACHE_TTL_SECONDS = 6 * 60 * 60
_project_version_cache = {"checked_at": 0.0, "latest_version": None}
_project_version_cache_lock = threading.Lock()
def _extract_project_version(text: str) -> str:
"""Extract the package version from a pyproject.toml document."""
match = re.search(r"(?m)^\s*version\s*=\s*[\"']([^\"']+)[\"']\s*$", str(text or ""))
return match.group(1).strip() if match else ""
def _get_local_project_version() -> str:
try:
with open(PROJECT_VERSION_FILE, encoding="utf-8") as version_file:
return _extract_project_version(version_file.read()) or "unknown"
except (OSError, UnicodeError):
return "unknown"
def _version_sort_key(version: str) -> tuple[int, ...]:
return tuple(int(part) for part in re.findall(r"\d+", str(version or "")))
def _get_project_version_info() -> dict[str, Any]:
"""Return the installed version and the version currently published on GitHub."""
current_version = _get_local_project_version()
now = time.monotonic()
with _project_version_cache_lock:
cached_at = float(_project_version_cache.get("checked_at") or 0.0)
latest_version = _project_version_cache.get("latest_version")
cache_is_fresh = now - cached_at < PROJECT_VERSION_CACHE_TTL_SECONDS
if not cache_is_fresh:
latest_version = None
try:
from .network_utils import request_source_response
response = request_source_response(
PROJECT_GITHUB_PYPROJECT_URL,
timeout=5,
max_attempts=1,
log_name="GitHub version check",
)
try:
if response is not None and response.status_code == 200:
latest_version = _extract_project_version(response.text)
finally:
if response is not None:
response.close()
except Exception:
latest_version = None
with _project_version_cache_lock:
_project_version_cache.update(
{
"checked_at": now,
"latest_version": latest_version or None,
}
)
current_key = _version_sort_key(current_version)
latest_key = _version_sort_key(latest_version or "")
if not latest_version or not latest_key:
status = "unavailable"
elif current_key and latest_key > current_key:
status = "update_available"
else:
status = "current"
return {
"current_version": current_version,
"latest_version": latest_version,
"status": status,
"github_url": PROJECT_GITHUB_URL,
}
+152 -18
View File
@@ -59,8 +59,145 @@ node_mod.extension.setup_routes()
extension = node_mod.extension
def _find_bound_extension(handler):
pending = [handler]
visited = set()
while pending:
current = pending.pop()
if id(current) in visited:
continue
visited.add(id(current))
for cell in getattr(current, "__closure__", ()) or ():
value = cell.cell_contents
if hasattr(value, "search_tracker"):
return value
if callable(value):
pending.append(value)
return None
class TestRefactoringTargets(unittest.IsolatedAsyncioTestCase):
def test_all_api_routes_are_registered(self):
expected_routes = {
("GET", "/model_resolver/base-models"),
("GET", "/model_resolver/base-models/status"),
("POST", "/model_resolver/base-models/update"),
("POST", "/model_resolver/analyze"),
("GET", "/model_resolver/analyze-progress/{analysis_id}"),
("POST", "/model_resolver/resolve"),
("POST", "/model_resolver/local-matches"),
("POST", "/model_resolver/local-model-hashes"),
("GET", "/model_resolver/model-preview"),
("POST", "/model_resolver/workflow-model-hashes"),
("POST", "/model_resolver/local-matches-by-hash"),
("POST", "/model_resolver/open-containing-folder"),
("POST", "/model_resolver/calculate-file-hash"),
("POST", "/model_resolver/calculate-file-hash/start"),
("GET", "/model_resolver/calculate-file-hash/progress/{progress_id}"),
("POST", "/model_resolver/calculate-file-hash/cancel/{progress_id}"),
("GET", "/model_resolver/models"),
("POST", "/model_resolver/metadata-size-audit"),
("GET", "/model_resolver/metadata-build/capabilities"),
("POST", "/model_resolver/metadata-build/start"),
("GET", "/model_resolver/metadata-build/progress/{progress_id}"),
("POST", "/model_resolver/metadata-build/cancel/{progress_id}"),
("POST", "/model_resolver/loaded"),
("GET", "/model_resolver/loaded-progress/{loaded_id}"),
("POST", "/model_resolver/civitai-search"),
("POST", "/model_resolver/custom-url"),
("POST", "/model_resolver/model-details"),
("GET", "/model_resolver/search-progress/{progress_id}"),
("POST", "/model_resolver/search-cancel/{progress_id}"),
("POST", "/model_resolver/search"),
("POST", "/model_resolver/clear-search-cache"),
("POST", "/model_resolver/civitai/session-token/check"),
("POST", "/model_resolver/civitai/api-key/check"),
("POST", "/model_resolver/huggingface/token/check"),
("POST", "/model_resolver/brave/api-key/check"),
("GET", "/model_resolver/huggingface/author-index/status"),
("POST", "/model_resolver/huggingface/author-index/refresh"),
("GET", "/model_resolver/model-list/status"),
("POST", "/model_resolver/model-list/update"),
("POST", "/model_resolver/download"),
("GET", "/model_resolver/progress/{download_id}"),
("GET", "/model_resolver/progress"),
("POST", "/model_resolver/cancel/{download_id}"),
("POST", "/model_resolver/pause/{download_id}"),
("POST", "/model_resolver/resume/{download_id}"),
("POST", "/model_resolver/clear_completed_downloads"),
("GET", "/model_resolver/aria2/status"),
("POST", "/model_resolver/aria2/status"),
("POST", "/model_resolver/aria2/start"),
("GET", "/model_resolver/aria2/stop"),
("POST", "/model_resolver/aria2/stop"),
("POST", "/model_resolver/aria2/install"),
("GET", "/model_resolver/directories"),
("GET", "/model_resolver/root-directories"),
("GET", "/model_resolver/path-template-suggestions"),
("GET", "/model_resolver/capabilities"),
("GET", "/model_resolver/version"),
("GET", "/model_resolver/subfolders/{category}"),
("GET", "/model_resolver/logs/backend/export"),
("GET", "/model_resolver/settings"),
("POST", "/model_resolver/settings"),
}
self.assertEqual(expected_routes, set(routes_registered))
def test_project_version_helpers_preserve_version_comparison_behavior(self):
self.assertEqual(
node_mod._extract_project_version('version = "1.2.3"'),
"1.2.3",
)
self.assertEqual(node_mod._extract_project_version("name = 'resolver'"), "")
self.assertEqual(node_mod._version_sort_key("v1.10.2-beta"), (1, 10, 2))
self.assertEqual(node_mod._version_sort_key("unknown"), ())
def test_project_version_info_reports_remote_update(self):
version_module = importlib.import_module("comfyui-model-resolver.core.version")
network_module = importlib.import_module("comfyui-model-resolver.core.network_utils")
response = MagicMock(status_code=200, text='version = "2.0.0"')
with (
patch.object(version_module, "_get_local_project_version", return_value="1.0.0"),
patch.object(network_module, "request_source_response", return_value=response),
):
version_module._project_version_cache.update(
{"checked_at": 0.0, "latest_version": None}
)
result = version_module._get_project_version_info()
self.assertEqual(result["current_version"], "1.0.0")
self.assertEqual(result["latest_version"], "2.0.0")
self.assertEqual(result["status"], "update_available")
self.assertEqual(result["github_url"], node_mod.PROJECT_GITHUB_URL)
async def test_version_route_returns_version_payload(self):
get_handler = routes_registered[("GET", "/model_resolver/version")]
request = MagicMock()
expected = {
"current_version": "1.0.0",
"latest_version": "1.1.0",
"status": "update_available",
}
version_module = importlib.import_module("comfyui-model-resolver.core.version")
with (
patch.object(version_module, "_get_project_version_info", return_value=expected),
patch("aiohttp.web.json_response") as mock_json_response,
):
await get_handler(request)
mock_json_response.assert_called_once_with(expected)
async def test_settings_route_returns_settings_and_schema(self):
get_handler = routes_registered[("GET", "/model_resolver/settings")]
with patch("aiohttp.web.json_response") as mock_json_response:
await get_handler(MagicMock())
mock_json_response.assert_called_once()
payload = mock_json_response.call_args.args[0]
self.assertIn("settings", payload)
self.assertIn("schema", payload)
@classmethod
def tearDownClass(cls):
@@ -188,15 +325,7 @@ class TestRefactoringTargets(unittest.IsolatedAsyncioTestCase):
get_handler = routes_registered.get(("GET", "/model_resolver/search-progress/{progress_id}"))
self.assertIsNotNone(get_handler)
bound_extension = None
if hasattr(get_handler, "__closure__") and get_handler.__closure__:
for cell in get_handler.__closure__:
val = cell.cell_contents
if hasattr(val, "search_tracker"):
bound_extension = val
break
target_extension = bound_extension or extension
target_extension = _find_bound_extension(get_handler) or extension
target_extension.search_tracker.update(progress_id, status="running", stage="civitai", message="Searching Civitai", percent=50)
mock_request = MagicMock()
@@ -210,21 +339,26 @@ class TestRefactoringTargets(unittest.IsolatedAsyncioTestCase):
self.assertEqual(response_data.get("status"), "running")
self.assertEqual(response_data.get("percent"), 50.0)
async def test_search_progress_route_rejects_missing_id(self):
get_handler = routes_registered[("GET", "/model_resolver/search-progress/{progress_id}")]
request = MagicMock()
request.match_info = {"progress_id": ""}
with patch("aiohttp.web.json_response") as mock_json_res:
await get_handler(request)
mock_json_res.assert_called_once_with(
{"error": "progress_id is required"},
status=400,
)
async def test_search_cancel_route(self):
progress_id = "test_cancel_job"
cancel_handler = routes_registered.get(("POST", "/model_resolver/search-cancel/{progress_id}"))
self.assertIsNotNone(cancel_handler)
bound_extension = None
if hasattr(cancel_handler, "__closure__") and cancel_handler.__closure__:
for cell in cancel_handler.__closure__:
val = cell.cell_contents
if hasattr(val, "search_tracker"):
bound_extension = val
break
target_extension = bound_extension or extension
target_extension = _find_bound_extension(cancel_handler) or extension
target_extension.search_tracker.update(progress_id, status="running")
mock_request = MagicMock()