Warning
This document is for an in-development version of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.
Source code for galaxy.tools.search
"""
Module for building and searching the index of installed tools.
Before changing index-building or searching related parts it is highly
recommended to read the docs at https://whoosh.readthedocs.io.
Schema - this is how we define the index, both for building and searching. A
field is created for each data element that we want to add e.g. tool name,
tool ID, description. The type of field and its attributes define how
entries for that field will be indexed and ultimately how they can be
searched. Score weighting (boost) is added here on a per-field bases, to
allow matches to important fields like "name" to receive a higher score.
Tokenizers - these take an attribute (e.g. name) and parse it into "tokens" to
be stored in the index. Can be done in many ways for different search
functionality. For example, the IDTokenizer creates one token for an entire
entry, resulting in an index field that requires a full-field match. The
default tokenizer will break an entry into words, so that single word
matches are possible.
Filters - various filters are available for processing content as the index is
built. A StopFilter removes common articles 'a', 'for', 'and' etc. A
StemmingFilter removes suffixes from words to create a 'base work' e.g.
stemming -> stem; opened -> open; philosophy -> philosoph.
"""
import logging
import os
import re
import shutil
import threading
from typing import (
Protocol,
runtime_checkable,
TYPE_CHECKING,
)
from whoosh import index
from whoosh.fields import Schema
from whoosh.index import LockError
from galaxy.config import GalaxyAppConfiguration
from galaxy.tools.source_store.search import (
build_search_document,
build_search_schema,
INDEX_WRITE_LOCK_TIMEOUT,
search_whoosh_index,
ToolSearchTuning,
ToolWhooshIndex,
)
from galaxy.util import ExecutionTimer
if TYPE_CHECKING:
from galaxy.tool_util.toolbox.views.interface import ToolPanelViewModel
from galaxy.tools import (
Tool,
ToolBox,
)
from galaxy.tools.cache import ToolCache
from galaxy.tools.source_store.index import ToolIndex
from galaxy.util.path import StrPath
log = logging.getLogger(__name__)
CanConvertToFloat = str | int | float
CanConvertToInt = str | int | float
UNSAFE_PATH_CHARACTERS = re.compile(r"[^A-Za-z0-9_-]")
[docs]
def panel_view_index_dir(index_dir: str, panel_view_id: str) -> str:
"""Directory holding the index of ``panel_view_id`` inside ``index_dir``.
Panel view ids such as ``ontology:edam_operations`` contain characters that
some filesystems (SMB/CIFS shares, NFS exports of NTFS-style volumes,
Windows) reject in file names, so every character outside
``[A-Za-z0-9_-]`` is percent-encoded.
"""
dir_name = UNSAFE_PATH_CHARACTERS.sub(
lambda match: "".join(f"%{byte:02X}" for byte in match.group().encode("utf-8")), panel_view_id
)
return os.path.join(index_dir, dir_name)
[docs]
def get_or_create_index(index_dir: "StrPath", schema: Schema) -> index.FileIndex:
"""Get or create a reference to the index."""
os.makedirs(index_dir, exist_ok=True)
if index.exists_in(index_dir):
idx = index.open_dir(index_dir)
if idx.schema == schema:
return idx
log.warning(f"Index at '{index_dir}' uses outdated schema, creating a new index")
# Delete the old index and return a new index reference
shutil.rmtree(index_dir)
os.makedirs(index_dir)
return index.create_in(index_dir, schema=schema)
[docs]
class ToolBoxSearch:
"""Support searching across all fixed panel views in a toolbox.
Search is delegated off to ToolPanelViewSearch for each panel object.
"""
[docs]
def __init__(self, toolbox: "ToolBox", index_dir: str, index_help: bool = True) -> None:
panel_searches: dict[str, ToolPanelViewSearch] = {}
for panel_view in toolbox.panel_views():
panel_view_id = panel_view.id
panel_index_dir = panel_view_index_dir(index_dir, panel_view_id)
panel_searches[panel_view_id] = ToolPanelViewSearch(
panel_view_id,
panel_index_dir,
index_help=index_help,
config=toolbox.app.config,
)
self.panel_searches = panel_searches
# ``reindex_tool_search`` reaches ``build_index`` concurrently from
# boot, the ``rebuild_toolbox_search_index`` control task and shed
# installs, all writing the same on-disk Whoosh dirs.
self._init_build_state()
def _init_build_state(self) -> None:
self._build_lock = threading.RLock()
self._index_built = threading.Condition(self._build_lock)
# The toolbox reload the index was last built from; None until the
# first build. Builds that are not reloads (shed installs, tool
# removals) record the same reload, so they never make a later
# reload look already indexed.
self.indexed_reload_count: int | None = None
def _mark_built(self, toolbox: "ToolBox | SupportsCachedSearch") -> None:
self.indexed_reload_count = toolbox._reload_count
self._index_built.notify_all()
[docs]
def build_index(self, tool_cache: "ToolCache", toolbox: "ToolBox", index_help: bool = True) -> None:
with self._build_lock:
for panel_search in self.panel_searches.values():
panel_search.build_index(tool_cache, toolbox, index_help=index_help)
self._mark_built(toolbox)
[docs]
def build_index_if_stale(self, tool_cache: "ToolCache", toolbox: "ToolBox", index_help: bool = True) -> bool:
"""Build unless the index already covers the toolbox's current reload.
The check runs under the build lock, so a caller arriving while a build
is in flight waits for it and then skips instead of building again.
"""
with self._build_lock:
if self.indexed_reload_count == toolbox._reload_count:
return False
self.build_index(tool_cache, toolbox, index_help=index_help)
return True
[docs]
def wait_until_current(self, toolbox: "ToolBox", timeout: float) -> bool:
"""Block until an index built from ``toolbox``'s current reload exists."""
with self._index_built:
return self._index_built.wait_for(lambda: self.indexed_reload_count == toolbox._reload_count, timeout)
[docs]
def search(self, q: str, panel_view: str, config: GalaxyAppConfiguration) -> list[str]:
if panel_view not in self.panel_searches:
raise KeyError(f"Unknown panel_view specified {panel_view}")
panel_search = self.panel_searches[panel_view]
return panel_search.search(q, config)
[docs]
@runtime_checkable
class SupportsCachedSearch(Protocol):
"""The toolbox surface :class:`CachedToolboxSearch` consumes."""
_reload_count: int
@property
def tool_index(self) -> "ToolIndex | None": ...
[docs]
class CachedToolboxSearch(ToolBoxSearch):
"""Build one metadata-only Whoosh corpus per rendered panel view."""
[docs]
def __init__(self, config: GalaxyAppConfiguration, toolbox: SupportsCachedSearch | None = None) -> None:
self.config = config
self._toolbox: SupportsCachedSearch | None = None
self.cached_panel_searches: dict[str, ToolWhooshIndex] = {}
self._panel_view_ids: set[str] = set()
# ``reindex_tool_search`` reaches ``build_index`` concurrently from
# boot, the ``rebuild_toolbox_search_index`` control task, and
# ``remove_tool_by_id`` — all writing the same on-disk Whoosh dirs.
# Two builds racing there corrupt each other: one thread's
# ``ToolWhooshIndex._open`` can ``create_in``/``rmtree`` a dir another
# thread is mid-write on, surfacing as ``whoosh.index.LockError`` or a
# fatal ``FileNotFoundError`` on the segment TOC (observed killing
# TestCachedDataManagerIntegration setup in CI). Serialize the whole
# build so only one rebuild touches the index dirs at a time.
self._init_build_state()
if toolbox is not None:
self._sync_panel_searches(toolbox)
[docs]
def build_index(self, tool_cache: "ToolCache", toolbox: "ToolBox", index_help: bool = True) -> None:
cached_toolbox = self._require_search_toolbox(toolbox)
with self._build_lock:
self._sync_panel_searches(cached_toolbox)
tool_index = cached_toolbox.tool_index
complete = True
if tool_index is not None:
for panel_view_id, searcher in self.cached_panel_searches.items():
try:
searcher.build(tool_index, cached_toolbox.panel_view_tool_ids(panel_view_id))
except LockError:
log.warning(
"Tool search index of panel %s is locked by another writer, skipping this rebuild",
panel_view_id,
)
complete = False
if complete:
self._mark_built(cached_toolbox)
else:
# The other writer may hold an older corpus: leave the index
# stale so the next rebuild_toolbox_search_index retries.
self.indexed_reload_count = None
[docs]
def search(self, q: str, panel_view: str, config: GalaxyAppConfiguration) -> list[str]:
if panel_view not in self._panel_view_ids:
raise KeyError(f"Unknown panel_view specified {panel_view}")
if not config.tool_search_index_dir:
return []
return self.cached_panel_searches[panel_view].search(q, limit=None)
def _sync_panel_searches(self, toolbox: SupportsCachedSearch) -> None:
self._toolbox = toolbox
panel_view_ids = {panel_view.id for panel_view in toolbox.panel_views()}
self._panel_view_ids = panel_view_ids
# Every build syncs, and each ``ToolWhooshIndex`` holds a pool of open
# searchers, so keep the ones whose directory and tuning still apply
# and close the others.
panel_searches = {}
if self.config.tool_search_index_dir:
tuning = ToolSearchTuning.from_config(self.config)
for panel_view_id in panel_view_ids:
index_dir = panel_view_index_dir(self.config.tool_search_index_dir, panel_view_id)
panel_search = self.cached_panel_searches.get(panel_view_id)
if panel_search is None or panel_search.index_dir != index_dir or panel_search.tuning != tuning:
panel_search = ToolWhooshIndex(index_dir=index_dir, tuning=tuning)
panel_searches[panel_view_id] = panel_search
for panel_search in self.cached_panel_searches.values():
if panel_search not in panel_searches.values():
panel_search.close()
self.cached_panel_searches = panel_searches
@staticmethod
def _require_search_toolbox(toolbox: "ToolBox") -> SupportsCachedSearch:
if not isinstance(toolbox, SupportsCachedSearch):
raise TypeError("CachedToolboxSearch requires a toolbox with a tool index, e.g. CachedToolBox")
return toolbox
[docs]
class ToolPanelViewSearch:
"""
Support searching tools in a toolbox. This implementation uses
the Whoosh search library.
"""
[docs]
def __init__(
self,
panel_view_id: str,
index_dir: str,
config: GalaxyAppConfiguration,
index_help: bool = True,
) -> None:
"""Build the schema and validate against the index."""
tuning = ToolSearchTuning.from_config(config)
self.schema = build_search_schema(
tuning,
help_boost=tuning.help_boost if index_help else None,
)
self.tuning = tuning
self.index_dir = index_dir
self.panel_view_id = panel_view_id
self.index = self._index_setup()
def _index_setup(self) -> index.FileIndex:
"""Get or create a reference to the index."""
return get_or_create_index(self.index_dir, self.schema)
[docs]
def build_index(self, tool_cache: "ToolCache", toolbox: "ToolBox", index_help: bool = True) -> None:
"""Prepare search index for tools loaded in toolbox.
Use `tool_cache` to determine which tools need indexing and which
should be removed.
"""
log.debug(f"Starting to build toolbox index of panel {self.panel_view_id}.")
execution_timer = ExecutionTimer()
with self.index.reader() as reader:
# Index ocasionally contains empty stored fields
self.indexed_tool_ids = {f["id"] for f in reader.all_stored_fields() if f}
tool_ids_to_remove = self._get_tools_to_remove(tool_cache)
tools_to_index = self._get_tool_list(
toolbox,
tool_cache,
)
# Commit synchronously so the index is complete when this returns and
# no writer thread outlives the call (or the app's shutdown).
try:
writer = self.index.writer(timeout=INDEX_WRITE_LOCK_TIMEOUT)
except LockError:
log.warning(
"Tool search index of panel %s is locked by another writer, skipping this rebuild",
self.panel_view_id,
)
return
with writer:
for tool_id in tool_ids_to_remove:
writer.delete_by_term("id", tool_id)
for tool in tools_to_index:
add_doc_kwds = self._create_doc(
tool=tool,
index_help=index_help,
)
# Add tool document to index (or overwrite if existing)
writer.update_document(**add_doc_kwds)
log.debug("Toolbox index of panel %s finished %s", self.panel_view_id, execution_timer)
def _get_tools_to_remove(self, tool_cache: "ToolCache") -> list[str]:
"""Return list of tool IDs to be removed from index."""
tool_ids_to_remove = (self.indexed_tool_ids - set(tool_cache._tool_paths_by_id.keys())).union(
tool_cache._removed_tool_ids
)
for indexed_tool_id in self.indexed_tool_ids:
indexed_tool = tool_cache.get_tool_by_id(indexed_tool_id)
if indexed_tool:
if indexed_tool.is_latest_version:
continue
latest_version = indexed_tool.latest_version
# `Tool.latest_version` resolves the newest revision through the
# tool cache under the lineage id `<versionless id>/<version>`.
# That key only exists for tools whose id carries their version,
# i.e. toolshed guids. A tool whose revisions share one bare id
# (`filters/grep.xml` and `filters/grep_1.0.1.xml` both declare
# `Grep1`) resolves to None, and the tool cache hands back
# whichever revision was parsed last. Both revisions are indexed
# under that one id, so an unresolvable latest version means the
# entry is still current, not that the tool went away.
if latest_version is None or latest_version.hidden:
continue
tool_ids_to_remove.add(indexed_tool_id)
return list(tool_ids_to_remove)
def _get_tool_list(self, toolbox: "ToolBox", tool_cache: "ToolCache") -> list["Tool"]:
"""Return list of tools to add and remove from index."""
tools_to_index: list[Tool] = []
for tool_id in tool_cache._new_tool_ids - self.indexed_tool_ids:
tool_like = toolbox.get_tool(tool_id)
tool = toolbox.materialize_tool(tool_like, reason="detail") if tool_like else None
if tool and tool.is_latest_version and toolbox.panel_has_tool(tool, self.panel_view_id):
if tool.hidden:
# Check if there is an older tool we can return
if tool.lineage:
tool_versions = reversed(tool.lineage.get_versions())
for tool_version in tool_versions:
tool = tool_cache.get_tool_by_id(tool_version.id)
if tool and not tool.hidden:
break
else:
continue
else:
continue
tools_to_index.append(tool)
return tools_to_index
def _create_doc(
self,
tool: "Tool",
index_help: bool = True,
) -> dict[str, str | list[str]]:
if tool.id is None:
return {}
document = build_search_document(
tool_id=tool.id,
guid=tool.guid,
name=tool.name,
description=tool.description,
section=tool.get_panel_section()[1],
edam_operations=tool.edam_operations,
edam_topics=tool.edam_topics,
repository=tool.repository_name,
owner=tool.repository_owner,
labels=tool.labels,
tool_tags=tool.tool_tags,
help_text=tool.raw_help if index_help else None,
tool_type=tool.tool_type,
)
return document or {}
[docs]
def search(
self,
q: str,
config: GalaxyAppConfiguration,
) -> list[str]:
"""Perform search on the in-memory index."""
tuning = ToolSearchTuning.from_config(config)
return [tool_id for tool_id, _score in search_whoosh_index(self.index, q, tuning, limit=None)]