mypy: fix type errors flagged by lint + packages jobs

- AbstractToolBox gains a no-op invalidate_index_cache (LazyToolBox
  overrides it); shed installs no longer depend on the subclass type.
- Annotate LazyToolboxSearch.build_index, ToolFileWatcher.observer,
  populator session cast, UUID key coercion in _register_lazy_entry,
  and drop stale type: ignore comments.
This commit is contained in:
mvdbeek
2026-07-28 17:27:22 +02:00
parent fb8b0b8f9e
commit 3eef867da8
8 changed files with 25 additions and 10 deletions
@@ -161,7 +161,9 @@ class ToolPanelManager:
populate_for_paths(
self.app.config,
self.app.model.context,
# Lazy mode implies a full Galaxy app, which carries
# ``model`` beyond the InstallationTarget protocol.
self.app.model.context, # type: ignore[attr-defined]
paths=new_paths,
rebuild_whoosh=True,
)
+1 -1
View File
@@ -134,7 +134,7 @@ def _resolve_file_template_kwds(root_dir: str | None) -> dict[str, str]:
"""
try:
# Lazy + optional: helper must still work outside a galaxy install.
from galaxy.tools import MODEL_TOOLS_PATH # type: ignore[attr-defined]
from galaxy.tools import MODEL_TOOLS_PATH
except Exception:
if root_dir:
MODEL_TOOLS_PATH = os.path.abspath(os.path.join(root_dir, "lib", "galaxy", "tools"))
+6 -2
View File
@@ -41,6 +41,7 @@ from datetime import (
from pathlib import Path
from typing import (
Any,
cast,
)
from galaxy.tool_source_store import (
@@ -65,6 +66,7 @@ from galaxy.config import GalaxyAppConfiguration
from galaxy.datatypes.registry import Registry
from galaxy.model import set_datatypes_registry
from galaxy.model.mapping import init_models_from_config
from galaxy.model.scoped_session import galaxy_scoped_session
from galaxy.queues import galaxy_exchange
from galaxy.tool_source_store import build_tool_source_store
from galaxy.tool_source_store.search import (
@@ -169,7 +171,9 @@ class ToolFileWatcher:
self.verbose = verbose
# Injected so tests can substitute a fake; default is the AMQP notifier.
self._notify = notify_callable or send_reload_notification
self.observer = None
# ``watchdog`` observer; typed Any because watchdog is an optional
# dependency imported inside :meth:`start`.
self.observer: Any = None
self._pending_changes: set[str] = set()
self._lock = threading.Lock()
self._debounce_timer: threading.Timer | None = None
@@ -862,7 +866,7 @@ def watch_mode(
log.info(f"Building tool source store (backend: {config.tool_source_store})...")
store = build_tool_source_store(config, model.context)
store = build_tool_source_store(config, cast("galaxy_scoped_session", model.context))
# Determine directories to watch from tool configurations
tools_dirs_set: set[Path] = set()
+8
View File
@@ -894,6 +894,14 @@ class AbstractToolBox(ManagesIntegratedToolPanelMixin):
return [tool]
return []
def invalidate_index_cache(self) -> None:
"""Drop any cached tool-index state.
No-op for the eager toolbox — there is no external index to go
stale. ``LazyToolBox`` overrides this to re-read the persistent
``ToolIndex`` after an out-of-band populator write.
"""
def tools(self):
# Snapshot the id list so a concurrent shed install can't mutate
# the dict mid-iteration.
+3 -3
View File
@@ -972,11 +972,11 @@ class LazyToolBox(ToolBox):
if not any(getattr(t, "id", None) == tool_id for t in bucket):
bucket.append(stub) # type: ignore[arg-type]
if entry.uuid:
self._tools_by_uuid[entry.uuid] = stub # type: ignore[assignment]
self._tools_by_uuid[UUID(entry.uuid)] = stub # type: ignore[assignment]
# Lineage: LazyLineageMap builds it from entries_by_version, which the
# populator wrote on the previous step, so .get() returns the right
# ToolLineage; register() is the fallback for an as-yet-unseen id.
stub._lineage = self._lineage_map.get(tool_id) or self._lineage_map.register(stub)
stub._lineage = self._lineage_map.get(tool_id) or self._lineage_map.register(stub) # type: ignore[arg-type]
# Place into the panel. The populator stamped panel_section_id /
# panel_section_name onto the entry; create the ToolSection if it
# doesn't already exist (peer-process install of the first tool in
@@ -990,7 +990,7 @@ class LazyToolBox(ToolBox):
self._tool_panel[section_key] = section
section.elems.append_tool(stub)
else:
self._tool_panel[f"tool_{tool_id}"] = stub # type: ignore[assignment]
self._tool_panel[f"tool_{tool_id}"] = stub
return stub
def _register_loaded_tool(self, tool: "Tool") -> None:
+2 -1
View File
@@ -30,6 +30,7 @@ import os
import re
import shutil
from typing import (
Any,
TYPE_CHECKING,
)
@@ -149,7 +150,7 @@ class LazyToolboxSearch(ToolBoxSearch):
self.panel_searches: dict[str, ToolPanelViewSearch] = {}
self.index_count = -1
def build_index(self, tool_cache, toolbox, index_help: bool = True) -> None:
def build_index(self, tool_cache: Any, toolbox: Any, index_help: bool = True) -> None:
# Populator side owns whoosh writes; bump the watermark so the
# rebuild_toolbox_search_index control task observes "in sync".
self.index_count += 1
+1 -1
View File
@@ -689,7 +689,7 @@ class ToolsService(ServiceBase):
for hit in self._search(query, view) or []:
try:
tool = self._get_tool(trans, hit, user=trans.user)
if tool:
if tool and tool.id:
results.append(tool.id)
except exceptions.AuthenticationFailed:
pass
+1 -1
View File
@@ -124,7 +124,7 @@ def test_to_panel_entry_does_not_materialise():
def test_to_dict_materialises():
calls = []
calls: list[Any] = []
class _Real:
def to_dict(self, trans, link_details, tool_help, **kw):