Load tools dynamically from database

lib/galaxy/app.py

@@ -358,7 +358,12 @@ def _configure_toolbox(self):
 
         self.dynamic_tool_manager = DynamicToolManager(self)
         self._toolbox_lock = threading.RLock()
-        self._toolbox = tools.ToolBox(self.config.tool_configs, self.config.tool_path, self)
+        if self.is_webapp:
+            self._toolbox = tools.ToolBox(self.config.tool_configs, self.config.tool_path, self)
+        else:
+            from galaxy.tools.toolbox.toolbox_adapter import DatabaseToolBox
+
+            self._toolbox = DatabaseToolBox(self.config.tool_configs, self.config.tool_path, self)
         galaxy_root_dir = os.path.abspath(self.config.root)
         file_path = os.path.abspath(self.config.file_path)
         app_info = AppInfo(

lib/galaxy/datatypes/registry.py

@@ -755,9 +755,8 @@ def load_external_metadata_tool(self, toolbox: "AbstractToolBox") -> None:
         # We need to be able to add a job to the queue to set metadata. The queue will currently only accept jobs with an associated
         # tool.  We'll load a special tool to be used for Auto-Detecting metadata; this is less than ideal, but effective
         # Properly building a tool without relying on parsing an XML file is near difficult...so we bundle with Galaxy.
-        set_meta_tool = toolbox.load_hidden_lib_tool(
-            os.path.abspath(os.path.join(os.path.dirname(__file__), "set_metadata_tool.xml"))
-        )
+        toolbox.load_hidden_lib_tool(os.path.abspath(os.path.join(os.path.dirname(__file__), "set_metadata_tool.xml")))
+        set_meta_tool = toolbox.get_tool("__SET_METADATA__")
         self.set_external_metadata_tool = cast("SetMetadataTool", set_meta_tool)
         self.log.debug("Loaded external metadata tool: %s", self.set_external_metadata_tool.id)
 

lib/galaxy/model/__init__.py

@@ -1392,6 +1392,17 @@ class ToolSource(Base, Dictifiable, RepresentById):
     id: Mapped[int] = mapped_column(primary_key=True)
     hash: Mapped[Optional[str]] = mapped_column(Unicode(255))
     source: Mapped[dict] = mapped_column(JSONType)
+    tool_source_class: Mapped[str]
+
+
+class GalaxyToolSourceAssociation(Base, RepresentById):
+    __tablename__ = "galaxy_tool_source_association"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    tool_id: Mapped[str] = mapped_column(Unicode(255))
+    tool_source_id: Mapped[int] = mapped_column(ForeignKey("tool_source.id"), index=True)
+    tool_source: Mapped[ToolSource] = relationship()
+    tool_dir: Mapped[Optional[str]] = mapped_column(Unicode(255))
 
 
 class ToolRequest(Base, Dictifiable, RepresentById):

lib/galaxy/model/migrations/alembic/versions_gxy/27044275d420_add_galaxy_tool_source_association_table.py

@@ -0,0 +1,48 @@
+"""Add GalaxyToolSourceAssociation table
+
+Revision ID: 27044275d420
+Revises: 1d1d7bf6ac02
+Create Date: 2025-11-11 13:47:57.306214
+
+"""
+
+from sqlalchemy import (
+    Column,
+    ForeignKey,
+    Integer,
+)
+
+from galaxy.model.custom_types import TrimmedString
+from galaxy.model.migrations.util import (
+    add_column,
+    create_table,
+    drop_column,
+    drop_table,
+    transaction,
+)
+
+# revision identifiers, used by Alembic.
+revision = "27044275d420"
+down_revision = "1d1d7bf6ac02"
+branch_labels = None
+depends_on = None
+
+galaxy_tool_source_association_table = "galaxy_tool_source_association"
+
+
+def upgrade():
+    with transaction():
+        create_table(
+            galaxy_tool_source_association_table,
+            Column("id", Integer, primary_key=True),
+            Column("tool_id", TrimmedString(255)),
+            Column("tool_dir", TrimmedString(255)),
+            Column("tool_source_id", Integer, ForeignKey("tool_source.id")),
+        )
+        add_column("tool_source", Column("tool_source_class", TrimmedString(255)))
+
+
+def downgrade():
+    with transaction():
+        drop_table(galaxy_tool_source_association_table)
+        drop_column("tool_source", "tool_source_class")

lib/galaxy/tool_util/toolbox/base.py

@@ -67,6 +67,7 @@
 if TYPE_CHECKING:
     from galaxy.model import (
         DynamicTool,
+        Job,
         User,
         Workflow,
     )
@@ -212,9 +213,10 @@ def __init__(
         self._tool_config_watcher = self.app.watchers.tool_config_watcher
         self._filter_factory = FilterFactory(self)
         self._tool_tag_manager = self.tool_tag_manager()
+        self.known_tool_source_hashes = self._get_toolbox_hashes()
         self._init_tools_from_configs(config_filenames)
 
-        if self.app.name == "galaxy" and self._integrated_tool_panel_config_has_contents:
+        if self.app.name == "galaxy" and self.app.is_webapp and self._integrated_tool_panel_config_has_contents:
             self._load_tool_panel()
 
         toolbox = self
@@ -258,11 +260,32 @@ def to_model(self) -> ToolPanelViewModel:
         for tool_panel_view in tool_panel_views_list:
             self._tool_panel_views[tool_panel_view.to_model().id] = tool_panel_view
 
-        if self.app.name == "galaxy":
+        if self.app.name == "galaxy" and self.app.is_webapp:
             self._load_tool_panel_views()
-        if save_integrated_tool_panel:
+        if self.app.is_webapp and save_integrated_tool_panel:
             self._save_integrated_tool_panel()
 
+    def tool_for_job(
+        self,
+        job: "Job",
+        exact=True,
+        check_access=True,
+        user: Optional["User"] = None,
+        tool_version: Optional[str] = None,
+    ) -> Optional["Tool"]:
+        if (dynamic_tool := job.dynamic_tool) is not None:
+            if check_access:
+                if not user:
+                    return None
+                if not dynamic_tool.public:
+                    self.app.dynamic_tool_manager.ensure_can_use_unprivileged_tool(user)
+            return self.dynamic_tool_to_tool(dynamic_tool)
+        else:
+            return self.get_tool(job.tool_id, tool_version=tool_version or job.tool_version, exact=exact)
+
+    def _get_toolbox_hashes(self):
+        return set()
+
     def _default_panel_view(self, trans):
         config = self.app.config
         if hasattr(config, "config_value_for_host"):

lib/galaxy/tools/__init__.py

@@ -50,13 +50,15 @@
 from galaxy.managers.credentials import build_credentials_context_response
 from galaxy.metadata import get_metadata_compute_strategy
 from galaxy.model import (
+    GalaxyToolSourceAssociation,
     History,
     HistoryDatasetAssociation,
     HistoryDatasetCollectionAssociation,
     Job,
     JobToOutputDatasetAssociation,
     StoredWorkflow,
     ToolRequest,
+    ToolSource as ModelToolSource,
 )
 from galaxy.model.dataset_collections.matching import MatchingCollections
 from galaxy.schema.credentials import CredentialsContext
@@ -552,6 +554,10 @@ def sa_session(self):
         """
         return self.app.model.context
 
+    def _get_toolbox_hashes(self):
+        stmt = select(model.ToolSource.hash).distinct()
+        return set(self.sa_session.execute(stmt).scalars())
+
     def reload_dependency_manager(self):
         self._init_dependency_manager()
 
@@ -598,7 +604,27 @@ def tools_by_id(self):
 
     def create_tool(self, config_file: StrPath, **kwds) -> "Tool":
         tool_source = self.get_expanded_tool_source(config_file)
-        return self._create_tool_from_source(tool_source, config_file=config_file, **kwds)
+        tool = self._create_tool_from_source(tool_source, config_file=config_file, **kwds)
+        if tool.app and tool.id:
+            from galaxy.util.hash_util import md5_hash_str
+
+            source_string = tool_source.to_string()
+            tool_source_hash = md5_hash_str(source_string)
+            if tool_source_hash not in self.known_tool_source_hashes:
+                model_tool_source = ModelToolSource()
+
+                model_tool_source.hash = tool_source_hash
+                model_tool_source.source = source_string
+                model_tool_source.tool_source_class = type(tool_source).__name__
+                gtsa = GalaxyToolSourceAssociation()
+                gtsa.tool_id = tool.id
+                gtsa.tool_source = model_tool_source
+                gtsa.tool_dir = tool.tool_dir
+                tool.app.model.session.add(model_tool_source)
+                tool.app.model.session.add(gtsa)
+                tool.app.model.session.commit()
+                self.known_tool_source_hashes.add(tool_source_hash)
+        return tool
 
     def get_expanded_tool_source(self, config_file: StrPath) -> ToolSource:
         try:

lib/galaxy/tools/special_tools.py

@@ -14,4 +14,5 @@
 def load_lib_tools(toolbox):
     for name, path in SPECIAL_TOOLS.items():
         tool = toolbox.load_hidden_lib_tool(os.path.abspath(os.path.join(os.path.dirname(__file__), path)))
-        log.debug("Loaded %s tool: %s", name, tool.id)
+        if tool:
+            log.debug("Loaded %s tool: %s", name, tool.id)

lib/galaxy/tools/toolbox/toolbox_adapter.py

@@ -0,0 +1,127 @@
+import os
+from os import PathLike
+from typing import (
+    List,
+    Literal,
+    Optional,
+    overload,
+    TYPE_CHECKING,
+    Union,
+)
+from uuid import UUID
+
+from sqlalchemy import select
+
+from galaxy.model import (
+    DynamicTool,
+    GalaxyToolSourceAssociation,
+    ToolSource,
+    User,
+)
+from galaxy.tool_util.deps import (
+    build_dependency_manager,
+    NullDependencyManager,
+)
+from galaxy.tool_util.toolbox import AbstractToolBox
+from galaxy.tools import (
+    create_tool_from_representation,
+    Tool,
+)
+
+
+class DatabaseToolBox(AbstractToolBox):
+
+    def __init__(
+        self,
+        config_filenames: List[str],
+        tool_root_dir,
+        app,
+        view_sources=None,
+        default_panel_view="default",
+        save_integrated_tool_panel: bool = True,
+    ) -> None:
+        super().__init__(
+            config_filenames, tool_root_dir, app, view_sources, default_panel_view, save_integrated_tool_panel
+        )
+        self._init_dependency_manager()
+
+    def load_tool(
+        self, config_file: str | PathLike[str], guid=None, tool_shed_repository=None, use_cached: bool = False, **kwds
+    ) -> Tool:
+        pass
+
+    def register_tool(self, tool: Tool) -> None:
+        pass
+
+    def create_dynamic_tool(self, dynamic_tool: DynamicTool) -> Tool:
+        return super().create_dynamic_tool(dynamic_tool)
+
+    def _init_integrated_tool_panel(self, config):
+        pass
+
+    def _init_tools_from_configs(self, config):
+        pass
+
+    def tool_tag_manager(self):
+        pass
+
+    def _init_dependency_manager(self):
+        use_tool_dependency_resolution = getattr(self.app, "use_tool_dependency_resolution", True)
+        if not use_tool_dependency_resolution:
+            self.dependency_manager = NullDependencyManager()
+            return
+        app_config_dict = self.app.config.config_dict
+        conf_file = app_config_dict.get("dependency_resolvers_config_file")
+        default_tool_dependency_dir = os.path.join(
+            self.app.config.data_dir, self.app.config.schema.defaults["tool_dependency_dir"]
+        )
+        self.dependency_manager = build_dependency_manager(
+            app_config_dict=app_config_dict,
+            conf_file=conf_file,
+            default_tool_dependency_dir=default_tool_dependency_dir,
+        )
+
+    @overload
+    def get_tool(
+        self,
+        tool_id: Optional[str] = None,
+        tool_version: Optional[str] = None,
+        tool_uuid: Optional[Union[UUID, str]] = None,
+        get_all_versions: Literal[False] = False,
+        exact: Optional[bool] = False,
+        user: Optional["User"] = None,
+    ) -> Optional["Tool"]: ...
+
+    @overload
+    def get_tool(
+        self,
+        tool_id: Optional[str] = None,
+        tool_version: Optional[str] = None,
+        tool_uuid: Optional[Union[UUID, str]] = None,
+        get_all_versions: Literal[True] = True,
+        exact: Optional[bool] = False,
+        user: Optional["User"] = None,
+    ) -> list["Tool"]: ...
+
+    def get_tool(
+        self,
+        tool_id: Optional[str] = None,
+        tool_version: Optional[str] = None,
+        tool_uuid: Optional[Union[UUID, str]] = None,
+        get_all_versions: Optional[bool] = False,
+        exact: Optional[bool] = False,
+        user: Optional["User"] = None,
+    ) -> Union[Optional["Tool"], list["Tool"]]:
+        if tool_id:
+            stmt = (
+                select(ToolSource.source, ToolSource.tool_source_class, GalaxyToolSourceAssociation.tool_dir)
+                .join(GalaxyToolSourceAssociation, ToolSource.id == GalaxyToolSourceAssociation.tool_source_id)
+                .where(GalaxyToolSourceAssociation.tool_id == tool_id)
+            )
+            row = self.app.model.session.execute(stmt).one_or_none()
+            if row:
+                source, tool_source_class, tool_dir = row
+                return create_tool_from_representation(
+                    app=self.app, raw_tool_source=source, tool_dir=tool_dir, tool_source_class=tool_source_class
+                )
+        return