Warning
This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.
Source code for galaxy.managers.tool_data
from pathlib import Path
from typing import (
Dict,
Optional,
)
from galaxy import exceptions
from galaxy.structured_app import StructuredApp
from galaxy.tools.data import (
TabularToolDataField,
TabularToolDataTable,
)
from galaxy.tools.data._schema import (
ToolDataDetails,
ToolDataEntryList,
ToolDataField,
)
[docs]class ToolDataManager:
"""
Interface/service object for interacting with tool data.
"""
@property
def data_tables(self) -> Dict[str, TabularToolDataTable]:
return self._app.tool_data_tables.data_tables
[docs] def index(self) -> ToolDataEntryList:
"""Return all tool data tables."""
data_tables = [table.to_dict() for table in self.data_tables.values()]
return ToolDataEntryList.parse_obj(data_tables)
[docs] def show(self, table_name: str) -> ToolDataDetails:
"""Get details of a given data table"""
data_table = self._data_table(table_name)
element_view = data_table.to_dict(view="element")
return ToolDataDetails.parse_obj(element_view)
[docs] def show_field(self, table_name: str, field_name: str) -> ToolDataField:
"""Get information about a partiular field in a tool data table"""
field = self._data_table_field(table_name, field_name)
return ToolDataField.parse_obj(field.to_dict())
[docs] def reload(self, table_name: str) -> ToolDataDetails:
"""Reloads a tool data table."""
data_table = self._data_table(table_name)
data_table.reload_from_files()
return self._reload_data_table(table_name)
[docs] def get_field_file_path(self, table_name: str, field_name: str, file_name: str) -> Path:
"""Get the absolute path to a given file name in the table field"""
field_value = self._data_table_field(table_name, field_name)
base_dir = Path(field_value.get_base_dir())
full_path = base_dir / file_name
if str(full_path) not in field_value.get_files():
raise exceptions.ObjectNotFound("No such path in data table field.")
return full_path.absolute()
[docs] def delete(self, table_name: str, values: Optional[str] = None) -> ToolDataDetails:
"""Removes an item from a data table"""
data_table = self._data_table(table_name)
if not values:
raise exceptions.RequestParameterInvalidException("Invalid values for data table item specified.")
split_values = values.split("\t")
if len(split_values) != len(data_table.get_column_name_list()):
raise exceptions.RequestParameterInvalidException(
f"Invalid data table item ( {values} ) specified. Wrong number of columns ({len(split_values)} given, {len(data_table.get_column_name_list())} required)."
)
data_table.remove_entry(split_values)
return self._reload_data_table(table_name)
def _data_table(self, table_name: str) -> TabularToolDataTable:
try:
return self.data_tables[table_name]
except KeyError:
raise exceptions.ObjectNotFound(f"No such data table {table_name}")
def _data_table_field(self, table_name: str, field_name: str) -> TabularToolDataField:
out = self._data_table(table_name).get_field(field_name)
if out is None:
raise exceptions.ObjectNotFound(f"No such field {field_name} in data table {table_name}.")
return out
def _reload_data_table(self, name: str) -> ToolDataDetails:
self._app.queue_worker.send_control_task("reload_tool_data_tables", noop_self=True, kwargs={"table_name": name})
return self.show(name)