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. """
[docs] def __init__(self, app: StructuredApp): self._app = app
@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, )