mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-21 05:45:37 +08:00
tool_util: fold id_util into galaxy.util.tool_version
extract_tool_id_from_xml/extract_tool_id_from_file/build_guid/ is_toolshed_guid had no callers. The one used function, extract_short_id_from_guid, is now short_tool_id next to remove_version_from_guid, and the manual rsplit spellings of the same extraction in the search-doc and populator ontology-id expansions go through the pair as well.
This commit is contained in:
mvdbeek
@@ -1,130 +0,0 @@
|
||||
"""
|
||||
Utilities for working with tool IDs and GUIDs.
|
||||
|
||||
Tool IDs come in several formats:
|
||||
- Short ID: Just the tool identifier (e.g., "bwa", "qiime2__feature_classifier__classify_sklearn")
|
||||
- GUID: Full toolshed identifier (e.g., "toolshed.g2.bx.psu.edu/repos/owner/name/tool_id/version")
|
||||
|
||||
This module provides utilities to parse and convert between these formats.
|
||||
"""
|
||||
|
||||
import re
|
||||
|
||||
|
||||
def extract_tool_id_from_xml(xml_content: str, max_read: int = 2000) -> str | None:
|
||||
"""
|
||||
Extract tool ID from XML content using a simple regex.
|
||||
|
||||
This is a fast extraction method that doesn't require full XML parsing.
|
||||
Useful when you need just the ID without macro expansion or validation.
|
||||
|
||||
Args:
|
||||
xml_content: The XML content to parse (can be partial).
|
||||
max_read: Maximum characters to read from content (default: 2000).
|
||||
|
||||
Returns:
|
||||
The tool ID if found, None otherwise.
|
||||
|
||||
Example:
|
||||
>>> extract_tool_id_from_xml('<tool id="bwa" version="1.0">')
|
||||
'bwa'
|
||||
"""
|
||||
content = xml_content[:max_read] if len(xml_content) > max_read else xml_content
|
||||
match = re.search(r'<tool[^>]+id=["\']([^"\']+)["\']', content)
|
||||
return match.group(1) if match else None
|
||||
|
||||
|
||||
def extract_tool_id_from_file(file_path: str, max_read: int = 2000) -> str | None:
|
||||
"""
|
||||
Extract tool ID from an XML file.
|
||||
|
||||
Args:
|
||||
file_path: Path to the tool XML file.
|
||||
max_read: Maximum characters to read from file (default: 2000).
|
||||
|
||||
Returns:
|
||||
The tool ID if found, None otherwise.
|
||||
"""
|
||||
try:
|
||||
with open(file_path) as f:
|
||||
content = f.read(max_read)
|
||||
return extract_tool_id_from_xml(content, max_read=max_read)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def extract_short_id_from_guid(guid: str) -> str | None:
|
||||
"""
|
||||
Extract the short tool ID from a toolshed GUID.
|
||||
|
||||
GUIDs have the format: toolshed.domain/repos/owner/name/tool_id/version
|
||||
This extracts just the tool_id part.
|
||||
|
||||
Args:
|
||||
guid: The full toolshed GUID.
|
||||
|
||||
Returns:
|
||||
The short tool ID, or the original guid if it's not in the expected format.
|
||||
|
||||
Examples:
|
||||
>>> extract_short_id_from_guid("toolshed.g2.bx.psu.edu/repos/devteam/bwa/bwa/0.1.0")
|
||||
'bwa'
|
||||
>>> extract_short_id_from_guid("toolshed.g2.bx.psu.edu/repos/iuc/qiime2__feature_classifier__classify_sklearn/qiime2__feature_classifier__classify_sklearn/2024.5.0+q2galaxy.2024.5.0")
|
||||
'qiime2__feature_classifier__classify_sklearn'
|
||||
>>> extract_short_id_from_guid("simple_tool")
|
||||
'simple_tool'
|
||||
"""
|
||||
if "/" not in guid:
|
||||
# Not a GUID, return as-is
|
||||
return guid
|
||||
|
||||
# Split on "/" and extract the second-to-last part
|
||||
# Format: toolshed.domain/repos/owner/name/tool_id/version
|
||||
# Parts: [domain, "repos", owner, name, tool_id, version]
|
||||
parts = guid.split("/")
|
||||
if len(parts) >= 2:
|
||||
# The tool_id is typically the second-to-last part (before version)
|
||||
tool_id = parts[-2]
|
||||
return tool_id
|
||||
|
||||
return guid
|
||||
|
||||
|
||||
def build_guid(tool_shed: str, owner: str, name: str, tool_id: str, version: str) -> str:
|
||||
"""
|
||||
Build a full toolshed GUID from components.
|
||||
|
||||
Args:
|
||||
tool_shed: The toolshed domain (e.g., "toolshed.g2.bx.psu.edu").
|
||||
owner: Repository owner.
|
||||
name: Repository name.
|
||||
tool_id: Tool ID.
|
||||
version: Tool version.
|
||||
|
||||
Returns:
|
||||
The full GUID string.
|
||||
|
||||
Example:
|
||||
>>> build_guid("toolshed.g2.bx.psu.edu", "devteam", "bwa", "bwa", "0.1.0")
|
||||
'toolshed.g2.bx.psu.edu/repos/devteam/bwa/bwa/0.1.0'
|
||||
"""
|
||||
return f"{tool_shed}/repos/{owner}/{name}/{tool_id}/{version}"
|
||||
|
||||
|
||||
def is_toolshed_guid(tool_id: str) -> bool:
|
||||
"""
|
||||
Check if a tool ID is a toolshed GUID.
|
||||
|
||||
Args:
|
||||
tool_id: The tool ID to check.
|
||||
|
||||
Returns:
|
||||
True if this looks like a toolshed GUID, False otherwise.
|
||||
|
||||
Examples:
|
||||
>>> is_toolshed_guid("toolshed.g2.bx.psu.edu/repos/devteam/bwa/bwa/0.1.0")
|
||||
True
|
||||
>>> is_toolshed_guid("bwa")
|
||||
False
|
||||
"""
|
||||
return "/repos/" in tool_id
|
||||
@@ -25,7 +25,6 @@ from galaxy.exceptions import (
|
||||
ObjectNotFound,
|
||||
RequestParameterInvalidException,
|
||||
)
|
||||
from galaxy.tool_util.id_util import extract_short_id_from_guid
|
||||
from galaxy.tool_util.ontologies.ontology_data import curated_tool_tags
|
||||
from galaxy.tool_util.parser import get_tool_source
|
||||
from galaxy.tool_util.toolbox.base import (
|
||||
@@ -55,7 +54,10 @@ from galaxy.tools.source_store.populator import (
|
||||
populate_store_inline,
|
||||
)
|
||||
from galaxy.util import listify
|
||||
from galaxy.util.tool_version import remove_version_from_guid
|
||||
from galaxy.util.tool_version import (
|
||||
remove_version_from_guid,
|
||||
short_tool_id,
|
||||
)
|
||||
from . import (
|
||||
create_tool_from_source,
|
||||
ToolBox,
|
||||
@@ -238,8 +240,7 @@ class LazyTool:
|
||||
def old_id(self) -> str:
|
||||
if "old_id" in self._overrides:
|
||||
return self._overrides["old_id"]
|
||||
short = extract_short_id_from_guid(self._entry.id)
|
||||
return short or self._entry.id
|
||||
return short_tool_id(self._entry.id)
|
||||
|
||||
@property
|
||||
def tool_tags(self) -> list[str]:
|
||||
@@ -917,8 +918,8 @@ class LazyToolBox(ToolBox):
|
||||
if self._tool_index is None:
|
||||
return
|
||||
for entry_id in self._tool_index.entries.keys():
|
||||
short_id = extract_short_id_from_guid(entry_id)
|
||||
if short_id and short_id != entry_id:
|
||||
short_id = short_tool_id(entry_id)
|
||||
if short_id != entry_id:
|
||||
self._shed_short_id_to_guids.setdefault(short_id, set()).add(entry_id)
|
||||
|
||||
# === Override get_tool for lazy loading ===
|
||||
@@ -1864,8 +1865,8 @@ class LazyToolBox(ToolBox):
|
||||
# resurrects the uninstalled tool via the eager get_tool
|
||||
# fall-through. Scrub every object belonging to this guid, but
|
||||
# leave sibling installs (other guids, other versions) alone.
|
||||
short_id = extract_short_id_from_guid(tool_id)
|
||||
if short_id and short_id != tool_id:
|
||||
short_id = short_tool_id(tool_id)
|
||||
if short_id != tool_id:
|
||||
bucket = self._tools_by_old_id.get(short_id)
|
||||
if bucket:
|
||||
survivors = [
|
||||
|
||||
@@ -94,6 +94,10 @@ from galaxy.tools.source_store.search import (
|
||||
from galaxy.util import listify
|
||||
from galaxy.util.hash_util import md5_hash_file
|
||||
from galaxy.util.properties import load_app_properties
|
||||
from galaxy.util.tool_version import (
|
||||
remove_version_from_guid,
|
||||
short_tool_id,
|
||||
)
|
||||
from galaxy.util.watcher import (
|
||||
EventHandler,
|
||||
get_observer_class,
|
||||
@@ -473,7 +477,7 @@ def build_index_entry_from_source(
|
||||
lowered = tool_id.lower()
|
||||
all_ids = [lowered]
|
||||
if "/repos/" in lowered:
|
||||
all_ids = [lowered, lowered.rsplit("/", 1)[0], lowered.rsplit("/", 2)[-2]]
|
||||
all_ids = [lowered, remove_version_from_guid(lowered) or lowered, short_tool_id(lowered)]
|
||||
# Same ontology expansion as ``Tool.__init__``: curated EDAM mapping
|
||||
# overrides and legacy bio.tools xrefs included.
|
||||
ontology_data = expand_ontology_data(tool_source, all_ids, biotools_metadata_source)
|
||||
|
||||
@@ -42,6 +42,10 @@ from galaxy.tools.source_store.index import (
|
||||
)
|
||||
from galaxy.util import unicodify
|
||||
from galaxy.util.hash_util import md5_hash_str
|
||||
from galaxy.util.tool_version import (
|
||||
remove_version_from_guid,
|
||||
short_tool_id,
|
||||
)
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
@@ -199,7 +203,7 @@ def _entry_to_doc(entry: ToolIndexEntry, *, include_help: bool = False) -> dict
|
||||
tool_id = (entry.id or "").lower()
|
||||
all_ids = [tool_id]
|
||||
if "/repos/" in tool_id:
|
||||
all_ids = [tool_id, tool_id.rsplit("/", 1)[0], tool_id.rsplit("/", 2)[-2]]
|
||||
all_ids = [tool_id, remove_version_from_guid(tool_id) or tool_id, short_tool_id(tool_id)]
|
||||
if tags := curated_tool_tags(all_ids):
|
||||
doc["tool_tags"] = unicodify(",".join(tags))
|
||||
if include_help and entry.help_text:
|
||||
|
||||
@@ -6,3 +6,15 @@ def remove_version_from_guid(guid: str) -> str | None:
|
||||
return None
|
||||
last_slash = guid.rfind("/")
|
||||
return guid[:last_slash]
|
||||
|
||||
|
||||
def short_tool_id(guid: str) -> str:
|
||||
"""
|
||||
Tool-id segment of a toolshed-derived tool_id(=guid),
|
||||
e.g. ``toolshed/repos/owner/name/tool_id/version`` -> ``tool_id``.
|
||||
Ids that aren't guids pass through unchanged.
|
||||
"""
|
||||
versionless = remove_version_from_guid(guid)
|
||||
if not versionless or "/" not in versionless:
|
||||
return guid
|
||||
return versionless.rsplit("/", 1)[-1]
|
||||
|
||||
Reference in New Issue
Block a user