From c6882cd926c6a7329ba355a5dc2583aa0410ca38 Mon Sep 17 00:00:00 2001 From: mvdbeek Date: Tue, 7 Jul 2026 14:45:40 +0200 Subject: [PATCH] api/sanitize_allow: narrow the response with a typed model MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Drop lib/galaxy/tools/source_store/models.py — 10 of its 11 pydantic models were orphaned scaffolding for a tool-source admin API that no longer exists (nothing referenced them). Keep the one that maps to a real endpoint, SanitizeAllowlistResponse, moved next to its only consumer and given a typed per-tool entry so it actually narrows the response. The controller now returns SanitizeAllowlistResponse(**...).model_dump(), which reproduces the exact JSON the client already consumes (verified round-trip-identical). Claude-Session: https://claude.ai/code/session_018L7ZmCv2ubKA3JNeSL8Pkr --- lib/galaxy/tools/source_store/models.py | 124 ------------------ .../webapps/galaxy/api/sanitize_allow.py | 28 +++- 2 files changed, 27 insertions(+), 125 deletions(-) delete mode 100644 lib/galaxy/tools/source_store/models.py diff --git a/lib/galaxy/tools/source_store/models.py b/lib/galaxy/tools/source_store/models.py deleted file mode 100644 index 7df4d379e7d..00000000000 --- a/lib/galaxy/tools/source_store/models.py +++ /dev/null @@ -1,124 +0,0 @@ -""" -Pydantic models for Tool Source Store API serialization. -""" - -from datetime import datetime -from typing import ( - Any, -) - -from pydantic import ( - BaseModel, - ConfigDict, - Field, -) - - -class ToolSourceResponse(BaseModel): - """Response model for tool source metadata.""" - - model_config = ConfigDict(from_attributes=True) - - hash: str = Field(description="Content hash (SHA256)") - tool_source_class: str = Field(description="Tool source class name") - tool_id: str | None = Field(None, description="Tool ID") - tool_version: str | None = Field(None, description="Tool version") - tool_dir: str | None = Field(None, description="Tool directory") - stored_at: datetime | None = Field(None, description="Storage timestamp") - - -class ToolSourceDetailResponse(ToolSourceResponse): - """Detailed response model including raw source.""" - - raw_source: str = Field(description="Raw tool source content") - metadata: dict[str, Any] | None = Field(None, description="Additional metadata") - - -class ToolSourceListResponse(BaseModel): - """Response model for listing tool sources.""" - - total_count: int = Field(description="Total number of tool sources") - items: list[ToolSourceResponse] = Field(description="List of tool sources") - - -class ToolSourceStatsResponse(BaseModel): - """Response model for tool source storage statistics.""" - - backend: str = Field(description="Storage backend type") - count: int = Field(description="Number of stored tool sources") - size_bytes: int | None = Field(None, description="Total storage size in bytes") - - -class ToolIndexEntryResponse(BaseModel): - """Response model for tool index entry.""" - - model_config = ConfigDict(from_attributes=True) - - id: str = Field(description="Tool ID") - uuid: str | None = Field(None, description="Tool UUID") - version: str | None = Field(None, description="Tool version") - name: str = Field(description="Tool name") - description: str = Field(description="Tool description") - panel_section_id: str | None = Field(None, description="Panel section ID") - panel_section_name: str | None = Field(None, description="Panel section name") - labels: list[str] = Field(default_factory=list, description="Tool labels") - edam_operations: list[str] = Field(default_factory=list, description="EDAM operations") - edam_topics: list[str] = Field(default_factory=list, description="EDAM topics") - hidden: bool = Field(False, description="Whether tool is hidden") - test_count: int = Field(0, description="Number of tests") - - -class ToolIndexStatsResponse(BaseModel): - """Response model for tool index statistics.""" - - index_size: int = Field(description="Number of tools in index") - memory_estimate_bytes: int = Field(description="Estimated memory usage") - version: str = Field(description="Index version") - built_at: datetime | None = Field(None, description="Index build timestamp") - - -class TestsSummaryResponse(BaseModel): - """Response model for /api/tools/tests_summary.""" - - # Dict of tool_id -> version -> {tool_name, count} - # Using Dict[str, Any] because nested structure - model_config = ConfigDict(extra="allow") - - -class RequirementResponse(BaseModel): - """Response model for a tool requirement.""" - - name: str = Field(description="Requirement name") - version: str | None = Field(None, description="Requirement version") - type: str = Field("package", description="Requirement type") - - -class SanitizeAllowlistResponse(BaseModel): - """Response model for /api/sanitize_allow.""" - - blocked_toolshed: list[dict[str, Any]] = Field(default_factory=list, description="Blocked tool shed tools") - allowed_toolshed: list[dict[str, Any]] = Field(default_factory=list, description="Allowed tool shed tools") - blocked_local: list[dict[str, Any]] = Field(default_factory=list, description="Blocked local tools") - allowed_local: list[dict[str, Any]] = Field(default_factory=list, description="Allowed local tools") - - -class CacheStatsResponse(BaseModel): - """Response model for cache statistics.""" - - tool_cache_size: int = Field(description="Number of cached Tool objects") - tool_cache_maxsize: int = Field(description="Maximum cache size") - index_size: int = Field(description="Number of tools in index") - index_memory_estimate: int = Field(description="Estimated index memory usage") - - -class ToolIndexEntryListResponse(BaseModel): - """Response model for paginated tool index entry listings.""" - - total_count: int = Field(description="Total number of index entries") - items: list[ToolIndexEntryResponse] = Field(description="List of index entries") - - -class ClearCacheResponse(BaseModel): - """Response model for cache clear operation.""" - - status: str = Field(description="Operation status") diff --git a/lib/galaxy/webapps/galaxy/api/sanitize_allow.py b/lib/galaxy/webapps/galaxy/api/sanitize_allow.py index 9f8145a16a7..cf3b155c0c5 100755 --- a/lib/galaxy/webapps/galaxy/api/sanitize_allow.py +++ b/lib/galaxy/webapps/galaxy/api/sanitize_allow.py @@ -7,6 +7,11 @@ from typing import ( Any, ) +from pydantic import ( + BaseModel, + Field, +) + from galaxy import web from galaxy.webapps.base.controller import BaseAPIController from galaxy.webapps.base.webapp import GalaxyWebTransaction @@ -14,6 +19,25 @@ from galaxy.webapps.base.webapp import GalaxyWebTransaction log = logging.getLogger(__name__) +class SanitizeAllowlistToolEntry(BaseModel): + """A single tool row in the /api/sanitize_allow response.""" + + tool_name: str = Field(description="Display name of the installed tool ('' if not installed)") + tool_id: list[str] = Field(description="Tool id split on '/'") + ids: dict[str, str] = Field(description="Derived id fragments (full/owner/repository/tool[/allowed])") + allowed: bool = Field(description="Whether the tool is on the sanitize allowlist") + toolshed: bool = Field(description="Whether the tool id is tool-shed-scoped") + + +class SanitizeAllowlistResponse(BaseModel): + """Response model for /api/sanitize_allow.""" + + blocked_toolshed: list[SanitizeAllowlistToolEntry] = Field(default_factory=list) + allowed_toolshed: list[SanitizeAllowlistToolEntry] = Field(default_factory=list) + blocked_local: list[SanitizeAllowlistToolEntry] = Field(default_factory=list) + allowed_local: list[SanitizeAllowlistToolEntry] = Field(default_factory=list) + + class SanitizeAllowController(BaseAPIController): @web.require_admin @web.expose_api @@ -102,4 +126,6 @@ class SanitizeAllowController(BaseAPIController): sanitize_dict["blocked_toolshed"].append(tool_dict) else: sanitize_dict["blocked_local"].append(tool_dict) - return sanitize_dict + # Route through the response model so the shape is validated/documented; + # model_dump reproduces the same JSON the client already consumes. + return SanitizeAllowlistResponse(**sanitize_dict).model_dump()