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()