Source code for galaxy.webapps.galaxy.api.dynamic_tools
import logging
from datetime import datetime
from typing import (
List,
Optional,
Union,
)
from pydantic import BaseModel
from galaxy.exceptions import ObjectNotFound
from galaxy.managers.context import (
ProvidesHistoryContext,
ProvidesUserContext,
)
from galaxy.managers.tools import (
DynamicToolManager,
tool_payload_to_tool,
)
from galaxy.model import User
from galaxy.schema.fields import (
DecodedDatabaseIdField,
EncodedDatabaseIdField,
)
from galaxy.tool_util.parameters import input_models_for_tool_source
from galaxy.tool_util.parameters.convert import cwl_runtime_model
from galaxy.tool_util.parser.yaml import YamlToolSource
from galaxy.tool_util_models import UserToolSource
from galaxy.tool_util_models.dynamic_tool_models import (
DynamicToolPayload,
DynamicUnprivilegedToolCreatePayload,
)
from galaxy.webapps.galaxy.api import (
depends,
DependsOnTrans,
DependsOnUser,
Router,
)
log = logging.getLogger(__name__)
router = Router(tags=["dynamic_tools"])
DatabaseIdOrUUID = Union[DecodedDatabaseIdField, str]
[docs]
class UnprivilegedToolResponse(BaseModel):
id: EncodedDatabaseIdField
uuid: str
active: bool
hidden: bool
tool_id: Optional[str]
tool_format: Optional[str]
create_time: datetime
representation: UserToolSource
[docs]
@router.cbv
class UnprivilegedToolsApi:
# Almost identical to dynamic tools api, but operates with tool ids
# and is scoped to to individual user and never adds to global toolbox
dynamic_tools_manager: DynamicToolManager = depends(DynamicToolManager)
[docs]
@router.get("/api/unprivileged_tools", response_model_exclude_defaults=True)
def index(self, active: bool = True, trans: ProvidesUserContext = DependsOnTrans) -> List[UnprivilegedToolResponse]:
if not trans.user:
return []
return [t.to_dict() for t in self.dynamic_tools_manager.list_unprivileged_tools(trans.user, active=active)]
[docs]
@router.get("/api/unprivileged_tools/{uuid}", response_model_exclude_defaults=True)
def show(self, uuid: str, user: User = DependsOnUser) -> UnprivilegedToolResponse:
dynamic_tool = self.dynamic_tools_manager.get_unprivileged_tool_by_uuid(user, uuid)
if dynamic_tool is None:
raise ObjectNotFound()
return UnprivilegedToolResponse(**dynamic_tool.to_dict())
[docs]
@router.post("/api/unprivileged_tools", response_model_exclude_defaults=True)
def create(
self, payload: DynamicUnprivilegedToolCreatePayload, user: User = DependsOnUser
) -> UnprivilegedToolResponse:
dynamic_tool = self.dynamic_tools_manager.create_unprivileged_tool(
user,
payload,
)
return UnprivilegedToolResponse(**dynamic_tool.to_dict())
[docs]
@router.post("/api/unprivileged_tools/build")
def build(
self,
payload: DynamicUnprivilegedToolCreatePayload,
history_id: DecodedDatabaseIdField,
trans: ProvidesHistoryContext = DependsOnTrans,
):
history = trans.app.history_manager.get_owned(history_id, trans.user)
tool = tool_payload_to_tool(trans.app, payload.representation.model_dump(by_alias=True))
if tool:
return tool.to_json(trans=trans, history=history or trans.history)
[docs]
@router.post("/api/unprivileged_tools/runtime_model")
def runtime_model(self, payload: DynamicUnprivilegedToolCreatePayload, user: User = DependsOnUser):
self.dynamic_tools_manager.ensure_can_use_unprivileged_tool(user)
represention = payload.representation.model_dump(by_alias=True)
tool_source = YamlToolSource(root_dict=represention)
input_bundle = input_models_for_tool_source(tool_source)
return cwl_runtime_model(input_bundle)
[docs]
@router.delete("/api/unprivileged_tools/{uuid}")
def delete(self, uuid: str, user: User = DependsOnUser):
"""
DELETE /api/unprivileged_tools/{encoded_dynamic_tool_id|tool_uuid}
Deactivate the specified dynamic tool. Deactivated tools will not
be loaded into the toolbox.
"""
dynamic_tool = self.dynamic_tools_manager.get_unprivileged_tool_by_uuid(user, uuid)
if not dynamic_tool:
raise ObjectNotFound()
self.dynamic_tools_manager.deactivate_unprivileged_tool(user, dynamic_tool)
[docs]
@router.cbv
class DynamicToolApi:
dynamic_tools_manager: DynamicToolManager = depends(DynamicToolManager)
[docs]
@router.get("/api/dynamic_tools", public=True)
def index(self):
return [t.to_dict() for t in self.dynamic_tools_manager.list_tools()]
[docs]
@router.get("/api/dynamic_tools/{dynamic_tool_id}", public=True)
def show(self, dynamic_tool_id: Union[DatabaseIdOrUUID, str]):
dynamic_tool = self.dynamic_tools_manager.get_tool_by_id_or_uuid(dynamic_tool_id)
if dynamic_tool is None:
raise ObjectNotFound()
return dynamic_tool.to_dict()
[docs]
@router.post("/api/dynamic_tools", require_admin=True)
def create(self, payload: DynamicToolPayload, trans: ProvidesUserContext = DependsOnTrans):
dynamic_tool = self.dynamic_tools_manager.create_tool(trans, payload, allow_load=payload.allow_load)
return dynamic_tool.to_dict()
[docs]
@router.delete("/api/dynamic_tools/{dynamic_tool_id}", require_admin=True)
def delete(self, dynamic_tool_id: DatabaseIdOrUUID):
"""
DELETE /api/dynamic_tools/{encoded_dynamic_tool_id|tool_uuid}
Deactivate the specified dynamic tool. Deactivated tools will not
be loaded into the toolbox.
"""
dynamic_tool = dynamic_tool = self.dynamic_tools_manager.get_tool_by_id_or_uuid(dynamic_tool_id)
updated_dynamic_tool = self.dynamic_tools_manager.deactivate(dynamic_tool)
return updated_dynamic_tool.to_dict()