Source code for galaxy.managers.tool_data
from pathlib import Path
from typing import (
cast,
Dict,
Optional,
)
from galaxy import exceptions
from galaxy.files import ConfiguredFileSources
from galaxy.files.uris import stream_url_to_file
from galaxy.model import DatasetInstance
from galaxy.structured_app import (
MinimalManagerApp,
StructuredApp,
)
from galaxy.tool_util.data import (
BundleProcessingOptions,
ToolDataTableManager,
)
from galaxy.tool_util.data._schema import (
ToolDataDetails,
ToolDataEntryList,
ToolDataField,
)
from galaxy.tools.data import (
TabularToolDataField,
TabularToolDataTable,
ToolDataTable,
)
[docs]class ToolDataManager:
"""
Interface/service object for interacting with tool data.
"""
@property
def data_tables(self) -> Dict[str, ToolDataTable]:
return self._app.tool_data_tables.data_tables
[docs] def index(self) -> ToolDataEntryList:
"""Return all tool data tables."""
return self._app.tool_data_tables.index()
[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.model_construct(**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.model_construct(**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._tabular_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 _tabular_data_table(self, table_name: str) -> TabularToolDataTable:
return cast(TabularToolDataTable, self._data_table(table_name))
def _data_table(self, table_name: str) -> ToolDataTable:
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._tabular_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)
[docs]class ToolDataImportManager:
file_sources: ConfiguredFileSources
tool_data_tables: ToolDataTableManager
[docs] def __init__(self, app: MinimalManagerApp):
self.file_sources = app.file_sources
self.tool_data_tables = app.tool_data_tables
[docs] def import_data_bundle_by_uri(self, config, uri: str, tool_data_file_path=None):
# an admin-only task - so allow file:// uris
if uri.startswith("file://"):
target = uri[len("file://") :]
else:
target = stream_url_to_file(
uri,
self.file_sources,
)
options = BundleProcessingOptions(
what="data import", # An alternative to this is sticking this in the bundle, only used for logging.
data_manager_path=config.galaxy_data_manager_data_path,
target_config_file=config.data_manager_config_file,
tool_data_file_path=tool_data_file_path,
)
self.tool_data_tables.import_bundle(
target,
options,
)
[docs] def import_data_bundle_by_dataset(self, config, dataset: DatasetInstance, tool_data_file_path=None):
options = BundleProcessingOptions(
what="data import", # An alternative to this is sticking this in the bundle, only used for logging.
data_manager_path=config.galaxy_data_manager_data_path,
target_config_file=config.data_manager_config_file,
tool_data_file_path=tool_data_file_path,
)
self.tool_data_tables.import_bundle(
dataset.extra_files_path,
options,
)