mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-21 05:45:37 +08:00
api/sanitize_allow: narrow the response with a typed model
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
This commit is contained in:
mvdbeek
@@ -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")
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user