opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-24 16:30:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

Data manager bundles.

Browse Source
This commit is contained in:
John Chilton
2022-12-20 10:35:17 -05:00
parent d484ec006b
commit 5c04e71a58
23 changed files with 1192 additions and 413 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/celery/tasks.py
+24
View File
@@ -31,6 +31,7 @@ from galaxy.managers.hdas import HDAManager
from galaxy.managers.lddas import LDDAManager
from galaxy.managers.markdown_util import generate_branded_pdf
from galaxy.managers.model_stores import ModelStoreManager
from galaxy.managers.tool_data import ToolDataImportManager
from galaxy.metadata.set_metadata import set_metadata_portable
from galaxy.model.scoped_session import galaxy_scoped_session
from galaxy.schema.tasks import (
@@ -358,6 +359,29 @@ def compute_dataset_hash(
dataset_manager.compute_hash(request)
@galaxy_task(action="import a data bundle")
def import_data_bundle(
hda_manager: HDAManager,
ldda_manager: LDDAManager,
tool_data_import_manager: ToolDataImportManager,
config: GalaxyAppConfiguration,
src: str,
uri: Optional[str] = None,
id: Optional[int] = None,
):
if src == "uri":
assert uri
tool_data_import_manager.import_data_bundle_by_uri(config, uri)
else:
assert id
dataset: model.DatasetInstance
if src == "hda":
dataset = hda_manager.by_id(id)
else:
dataset = ldda_manager.by_id(id)
tool_data_import_manager.import_data_bundle_by_dataset(config, dataset)
@galaxy_task(action="pruning history audit table")
def prune_history_audit_table(sa_session: galaxy_scoped_session):
"""Prune ever growing history_audit table."""
lib/galaxy/datatypes/data.py
+1
View File
@@ -452,6 +452,7 @@ class Data(metaclass=DataMeta):
# Relocate all composite datatype display to a common location.
composite_extensions = trans.app.datatypes_registry.get_composite_extensions()
composite_extensions.append("html") # for archiving composite datatypes
composite_extensions.append("data_manager_json") # for downloading bundles if bundled.
# Prevent IE8 from sniffing content type since we're explicit about it. This prevents intentionally text/plain
# content from being rendered in the browser
headers["X-Content-Type-Options"] = "nosniff"
lib/galaxy/managers/tool_data.py
+48 -1
View File
@@ -6,7 +6,14 @@ from typing import (
)
from galaxy import exceptions
from galaxy.structured_app import StructuredApp
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
from galaxy.tool_util.data._schema import (
ToolDataDetails,
ToolDataEntryList,
@@ -16,6 +23,7 @@ from galaxy.tools.data import (
TabularToolDataField,
TabularToolDataTable,
ToolDataTable,
ToolDataTableManager,
)
@@ -95,3 +103,42 @@ class ToolDataManager:
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)
class ToolDataImportManager:
file_sources: ConfiguredFileSources
tool_data_tables: ToolDataTableManager
def __init__(self, app: MinimalManagerApp):
self.file_sources = app.file_sources
self.tool_data_tables = app.tool_data_tables
def import_data_bundle_by_uri(self, config, uri: str):
# 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,
)
self.tool_data_tables.import_bundle(
target,
options,
)
def import_data_bundle_by_dataset(self, config, dataset: DatasetInstance):
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,
)
self.tool_data_tables.import_bundle(
dataset.extra_files_path,
options,
)
lib/galaxy/schema/schema.py
+18
View File
@@ -2273,6 +2273,24 @@ class GroupUserListModel(Model):
__root__: List[GroupUserModel]
class ImportToolDataBundleUriSource(Model):
src: Literal["uri"] = Field(title="src", description="Indicates that the tool data should be resolved by a URI.")
uri: str = Field(
title="uri",
description="URI to fetch tool data bundle from (file:// URIs are fine because this is an admin-only operation)",
)
class ImportToolDataBundleDatasetSource(Model):
src: Literal["hda", "ldda"] = Field(
title="src", description="Indicates that the tool data should be resolved from a dataset."
)
id: DecodedDatabaseIdField = EntityIdField
ImportToolDataBundleSource = Union[ImportToolDataBundleDatasetSource, ImportToolDataBundleUriSource]
class ToolShedRepository(Model):
tool_shed_url: str = Field(
title="Tool Shed URL", default="https://toolshed.g2.bx.psu.edu/", description="Tool Shed target"
lib/galaxy/structured_app.py
+1
View File
@@ -97,6 +97,7 @@ class MinimalManagerApp(MinimalApp):
file_sources: ConfiguredFileSources
genome_builds: GenomeBuilds
dataset_collection_manager: "DatasetCollectionManager"
tool_data_tables: "ToolDataTableManager"
history_manager: "HistoryManager"
hda_manager: "HDAManager"
workflow_manager: "WorkflowsManager"
lib/galaxy/tool_util/data/__init__.py
+328 -10
View File
@@ -15,6 +15,7 @@ import os.path
import re
import string
import time
from dataclasses import dataclass
from glob import glob
from tempfile import NamedTemporaryFile
from typing import (
@@ -27,7 +28,6 @@ from typing import (
Set,
Tuple,
Type,
TYPE_CHECKING,
Union,
)
@@ -43,20 +43,25 @@ from galaxy.util import (
Element,
RW_R__R__,
)
from galaxy.util.compression_utils import decompress_path_to_directory
from galaxy.util.dictifiable import Dictifiable
from galaxy.util.filelock import FileLock
from galaxy.util.path import StrPath
from galaxy.util.renamed_temporary_file import RenamedTemporaryFile
from galaxy.util.template import fill_template
from ._schema import (
ToolDataEntry,
ToolDataEntryList,
)
if TYPE_CHECKING:
from galaxy.tools.data_manager.manager import DataManager
from .bundles.models import (
DataTableBundle,
DataTableBundleProcessorDescription,
RepoInfo,
)
log = logging.getLogger(__name__)
BUNDLE_INDEX_FILE_NAME = "_gx_data_bundle_index.json"
DEFAULT_TABLE_TYPE = "tabular"
TOOL_DATA_TABLE_CONF_XML = """<?xml version="1.0"?>
@@ -194,7 +199,7 @@ class ToolDataTable(Dictifiable):
entry: Union[List[str], Dict[str, str]],
allow_duplicates: bool = True,
persist: bool = False,
entry_source=None,
entry_source: Optional[Union[dict, RepoInfo]] = None,
**kwd,
) -> None:
raise NotImplementedError("Abstract method")
@@ -204,14 +209,19 @@ class ToolDataTable(Dictifiable):
entry: Union[List[str], Dict[str, str]],
allow_duplicates: bool = True,
persist: bool = False,
entry_source=None,
entry_source: Optional[Union[dict, RepoInfo]] = None,
**kwd,
) -> int:
self._add_entry(entry, allow_duplicates=allow_duplicates, persist=persist, entry_source=entry_source, **kwd)
return self._update_version()
def add_entries(
self, entries: List[List[str]], allow_duplicates: bool = True, persist: bool = False, entry_source=None, **kwd
self,
entries: List[List[str]],
allow_duplicates: bool = True,
persist: bool = False,
entry_source: Optional[Union[dict, RepoInfo]] = None,
**kwd,
) -> int:
for entry in entries:
try:
@@ -596,7 +606,7 @@ class TabularToolDataTable(ToolDataTable):
# This method is used in tools, so need to keep its API stable
def get_filename_for_source(
self, source: Optional[Union[Dict, "DataManager"]], default: Optional[str] = None
self, source: Optional[Union[Dict, RepoInfo]], default: Optional[str] = None
) -> Optional[str]:
source_repo_info: Optional[dict] = None
if source:
@@ -604,7 +614,7 @@ class TabularToolDataTable(ToolDataTable):
if isinstance(source, dict):
source_repo_info = source
else:
source_repo_info = source.get_tool_shed_repository_info_dict()
source_repo_info = source.dict()
filename = default
for name, value in self.filenames.items():
repo_info = value.get("tool_shed_repository")
@@ -620,7 +630,7 @@ class TabularToolDataTable(ToolDataTable):
entry: Union[List[str], Dict[str, str]],
allow_duplicates: bool = True,
persist: bool = False,
entry_source=None,
entry_source: Optional[Union[dict, RepoInfo]] = None,
**kwd,
) -> None:
# accepts dict or list of columns
@@ -827,6 +837,26 @@ def _expand_here_template(content: str, here: Optional[str]) -> str:
tool_data_table_types_list: List[Type[ToolDataTable]] = [TabularToolDataTable]
class HasExtraFiles(Protocol):
extra_files_path: str
def extra_files_path_exists(self) -> bool:
...
class DirectoryAsExtraFiles(HasExtraFiles):
def __init__(self, directory):
self.extra_files_path = directory
def extra_files_path_exists(self) -> bool:
return True
class OutputDataset(HasExtraFiles):
ext: str
file_name: str
class ToolDataTableManager(Dictifiable):
"""Manages a collection of tool data tables"""
@@ -1070,3 +1100,291 @@ class ToolDataTableManager(Dictifiable):
if path in data_table.filenames:
table_names.add(name)
return list(table_names)
def process_bundle(
self,
out_data: Dict[str, OutputDataset],
bundle_description: DataTableBundleProcessorDescription,
repo_info: Optional[RepoInfo],
options: "BundleProcessingOptions",
) -> List[str]:
data_manager_dict: Dict[str, Any] = _data_manager_dict(out_data)
bundle = DataTableBundle(
processor_description=bundle_description,
data_tables=data_manager_dict.get("data_tables", {}),
repo_info=repo_info,
)
return _process_bundle(out_data, bundle, options, self)
def import_bundle(
self,
target: str,
options: "BundleProcessingOptions",
) -> List[str]:
if not os.path.isdir(target):
target_directory = decompress_path_to_directory(target)
else:
target_directory = target
index_json = os.path.join(target_directory, BUNDLE_INDEX_FILE_NAME)
with open(index_json, "r") as f:
index = json.load(f)
bundle = DataTableBundle(**index)
assert bundle.output_name
out_data = {bundle.output_name: DirectoryAsExtraFiles(target_directory)}
return _process_bundle(out_data, bundle, options, self)
def write_bundle(
self,
out_data: Dict[str, OutputDataset],
bundle_description: DataTableBundleProcessorDescription,
repo_info: Optional[RepoInfo],
) -> None:
data_manager_dict = _data_manager_dict(out_data, ensure_single_output=True)
for output_name, dataset in out_data.items():
if dataset.ext != "data_manager_json":
continue
bundle = DataTableBundle(
data_tables=data_manager_dict.get("data_tables", {}),
output_name=output_name,
processor_description=bundle_description,
repo_info=repo_info,
)
extra_files_path = dataset.extra_files_path
bundle_path = os.path.join(extra_files_path, BUNDLE_INDEX_FILE_NAME)
with open(bundle_path, "w") as fw:
json.dump(bundle.dict(), fw)
SUPPORTED_DATA_TABLE_TYPES = TabularToolDataTable
@dataclass
class BundleProcessingOptions:
what: str
data_manager_path: str
target_config_file: str
def _data_manager_dict(out_data: Dict[str, OutputDataset], ensure_single_output: bool = False) -> Dict[str, Any]:
data_manager_dict: Dict[str, Any] = {}
found_output = False
for output_name, output_dataset in out_data.items():
if output_dataset.ext != "data_manager_json":
continue
if found_output and ensure_single_output:
raise Exception("Galaxy can only write bundles for data managers with a single output data_manager_json.")
found_output = True
try:
output_dict = json.loads(open(output_dataset.file_name).read())
except Exception as e:
log.warning(f'Error reading DataManagerTool json for "{output_name}": {e}')
continue
for key, value in output_dict.items():
if key not in data_manager_dict:
data_manager_dict[key] = {}
data_manager_dict[key].update(value)
data_manager_dict.update(output_dict)
return data_manager_dict
from typing import Mapping
def _process_bundle(
out_data: Mapping[str, HasExtraFiles],
bundle: DataTableBundle,
options: BundleProcessingOptions,
tool_data_tables: ToolDataTableManager,
):
updated_data_tables = []
data_tables_dict = bundle.data_tables
bundle_description = bundle.processor_description
for data_table_name in bundle_description.data_table_names:
data_table_values = data_tables_dict.pop(data_table_name, None)
if not data_table_values:
log.warning(f'No values for data table "{data_table_name}" were returned by "{options.what}".')
continue # next data table
data_table_remove_values = None
if isinstance(data_table_values, dict):
values_to_add = data_table_values.get("add")
data_table_remove_values = data_table_values.get("remove")
if values_to_add or data_table_remove_values:
# We don't have an old style data table definition
data_table_values = values_to_add
data_table = tool_data_tables.get(data_table_name, None)
if data_table is None:
log.error(
f'Processing by {options.what} returned an unknown data table "{data_table_name}" with new entries "{data_table_values}". These entries will not be created. Please confirm that an entry for "{data_table_name}" exists in your "tool_data_table_conf.xml" file.'
)
continue # next table name
if not isinstance(data_table, SUPPORTED_DATA_TABLE_TYPES):
log.error(
f'Processing by {options.what} returned an unsupported data table "{data_table_name}" with type "{type(data_table)}" with new entries "{data_table_values}". These entries will not be created. Please confirm that the data table is of a supported type ({SUPPORTED_DATA_TABLE_TYPES}).'
)
continue # next table name
output_ref_values = {}
output_ref_by_data_table = bundle_description.output_ref_by_data_table
if data_table_name in output_ref_by_data_table:
for data_table_column, output_ref in output_ref_by_data_table[data_table_name].items():
output_ref_dataset = out_data.get(output_ref, None)
assert output_ref_dataset is not None, "Referenced output was not found."
output_ref_values[data_table_column] = output_ref_dataset
if not isinstance(data_table_values, list):
data_table_values = [data_table_values] if data_table_values else []
if not isinstance(data_table_remove_values, list):
data_table_remove_values = [data_table_remove_values] if data_table_remove_values else []
for data_table_row in data_table_values:
data_table_value = dict(**data_table_row) # keep original values here
for (
name
) in (
data_table_row.keys()
): # FIXME: need to loop through here based upon order listed in data_manager config
if name in output_ref_values:
_process_move(
data_table_name,
name,
output_ref_values[name].extra_files_path,
bundle_description,
options,
**data_table_value,
)
data_table_value[name] = _process_value_translations(
data_table_name, name, bundle_description, options, **data_table_value
)
data_table.add_entry(data_table_value, persist=True, entry_source=bundle.repo_info)
# Removes data table entries
for data_table_row in data_table_remove_values:
data_table_value = dict(**data_table_row) # keep original values here
data_table.remove_entry(list(data_table_value.values()))
updated_data_tables.append(data_table_name)
if bundle_description.undeclared_tables and data_tables_dict:
# We handle the data move, by just moving all the data out of the extra files path
# moving a directory and the target already exists, we move the contents instead
log.debug("Attempting to add entries for undeclared tables: %s.", ", ".join(data_tables_dict.keys()))
for ref_file in out_data.values():
if ref_file.extra_files_path_exists():
util.move_merge(ref_file.extra_files_path, options.data_manager_path)
path_column_names = ["path"]
for data_table_name, data_table_values in data_tables_dict.items():
data_table = tool_data_tables.get(data_table_name, None)
if not isinstance(data_table_values, list):
data_table_values = [data_table_values]
for data_table_row in data_table_values:
data_table_value = dict(**data_table_row) # keep original values here
for name, value in data_table_row.items():
if name in path_column_names:
data_table_value[name] = os.path.abspath(os.path.join(options.data_manager_path, value))
data_table.add_entry(data_table_value, persist=True, entry_source=bundle.repo_info)
updated_data_tables.append(data_table_name)
else:
for data_table_name, data_table_values in data_tables_dict.items():
# tool returned extra data table entries, but data table was not declared in data manager
# do not add these values, but do provide messages
log.warning(
f'Processing by {options.what} returned an undeclared data table "{data_table_name}" with new entries "{data_table_values}". These entries will not be created. Please confirm that an entry for "{data_table_name}" exists in your "{options.target_config_file}" file.'
)
return updated_data_tables
def _process_move(
data_table_name: str,
column_name: str,
source_base_path: str,
bundle_description: DataTableBundleProcessorDescription,
options: BundleProcessingOptions,
**kwd,
):
move_by_data_table_column = bundle_description.move_by_data_table_column
if data_table_name in move_by_data_table_column and column_name in move_by_data_table_column[data_table_name]:
move = move_by_data_table_column[data_table_name][column_name]
source = move.source_base
if source is None:
source = source_base_path
else:
source = fill_template(
source,
GALAXY_DATA_MANAGER_DATA_PATH=options.data_manager_path,
**kwd,
).strip()
assert source
if move.source_value:
source = os.path.join(
source,
fill_template(
move.source_value,
GALAXY_DATA_MANAGER_DATA_PATH=options.data_manager_path,
**kwd,
).strip(),
)
target = move.target_base
if target is None:
target = options.data_manager_path
else:
target = fill_template(
target,
GALAXY_DATA_MANAGER_DATA_PATH=options.data_manager_path,
**kwd,
).strip()
assert target
if move.target_value:
target = os.path.join(
target,
fill_template(
move.target_value,
GALAXY_DATA_MANAGER_DATA_PATH=options.data_manager_path,
**kwd,
).strip(),
)
if move.type == "file":
dirs = os.path.split(target)[0]
try:
os.makedirs(dirs)
except OSError as e:
if e.errno != errno.EEXIST:
raise e
# moving a directory and the target already exists, we move the contents instead
if os.path.exists(source):
util.move_merge(source, target)
if move.relativize_symlinks:
util.relativize_symlinks(target)
return True
return False
def _process_value_translations(
data_table_name: str,
column_name: str,
bundle_description: DataTableBundleProcessorDescription,
options: BundleProcessingOptions,
**kwd,
) -> str:
value_translation_by_data_table_column = bundle_description.value_translation_by_data_table_column
value = kwd.get(column_name)
if (
data_table_name in value_translation_by_data_table_column
and column_name in value_translation_by_data_table_column[data_table_name]
):
for value_translation in value_translation_by_data_table_column[data_table_name][column_name]:
if isinstance(value_translation, str):
value = fill_template(
value_translation,
GALAXY_DATA_MANAGER_DATA_PATH=options.data_manager_path,
**kwd,
).strip()
else:
value = value_translation(value)
assert value
return value
lib/galaxy/tool_util/data/bundles/models.py
+270
View File
@@ -0,0 +1,270 @@
import os
from typing import (
Callable,
Dict,
Iterator,
List,
Optional,
Tuple,
Union,
)
from pydantic import (
BaseModel,
Extra,
root_validator,
)
from galaxy.util import (
asbool,
Element,
)
ALLOW_EXTRA = Extra.forbid
DEFAULT_VALUE_TRANSLATION_TYPE = "template"
VALUE_TRANSLATION_FUNCTIONS: Dict[str, Callable] = dict(abspath=os.path.abspath)
DEFAULT_VALUE_TRANSLATION_TYPE = "template"
class DataTableBundleProcessorDataTableOutputColumnTranslation(BaseModel):
type: str
value: str
class Config:
extra = ALLOW_EXTRA
class DataTableBundleProcessorDataTableOutputColumnMove(BaseModel):
type: str
source_base: Optional[str] = None
source_value: str = ""
target_base: Optional[str] = None
target_value: Optional[str] = None
relativize_symlinks: bool
class Config:
extra = ALLOW_EXTRA
class DataTableBundleProcessorDataTableOutputColumn(BaseModel):
name: str
data_table_name: str
output_ref: Optional[str] = None
value_translations: List[DataTableBundleProcessorDataTableOutputColumnTranslation] = []
moves: List[DataTableBundleProcessorDataTableOutputColumnMove] = []
class Config:
extra = ALLOW_EXTRA
@root_validator(pre=True)
def fill_in_default_data_table_name(cls, values):
data_table_name = values.get("data_table_name")
if data_table_name is None:
values["data_table_name"] = values["name"]
return values
class DataTableBundleProcessorDataTableOutput(BaseModel):
columns: List[DataTableBundleProcessorDataTableOutputColumn]
class Config:
extra = ALLOW_EXTRA
class DataTableBundleProcessorDataTable(BaseModel):
name: str
output: Optional[DataTableBundleProcessorDataTableOutput]
class Config:
extra = ALLOW_EXTRA
class DataTableBundleProcessorDescription(BaseModel):
undeclared_tables: bool = False
data_tables: List[DataTableBundleProcessorDataTable]
class Config:
extra = ALLOW_EXTRA
@property
def data_table_names(self) -> List[str]:
names = []
for data_table in self.data_tables:
data_table_name = data_table.name
names.append(data_table_name)
return names
def _walk_columns(self) -> Iterator[Tuple[str, DataTableBundleProcessorDataTableOutputColumn]]:
for data_table in self.data_tables:
data_table_name = data_table.name
output = data_table.output
if output:
for column in output.columns:
yield (data_table_name, column)
@property
def output_ref_by_data_table(self) -> Dict[str, Dict[str, str]]:
output_refs: Dict[str, Dict[str, str]] = {}
for data_table_name, column in self._walk_columns():
data_table_column_name = column.data_table_name
output_ref = column.output_ref
if output_ref is not None:
if data_table_name not in output_refs:
output_refs[data_table_name] = {}
output_refs[data_table_name][data_table_column_name] = output_ref
return output_refs
@property
def move_by_data_table_column(self) -> Dict[str, Dict[str, DataTableBundleProcessorDataTableOutputColumnMove]]:
by_column: Dict[str, Dict[str, DataTableBundleProcessorDataTableOutputColumnMove]] = {}
for data_table_name, column in self._walk_columns():
data_table_column_name = column.data_table_name
for move in column.moves:
if data_table_name not in by_column:
by_column[data_table_name] = {}
by_column[data_table_name][data_table_column_name] = move
return by_column
@property
def value_translation_by_data_table_column(self) -> Dict[str, Dict[str, List[Union[str, Callable]]]]:
by_column: Dict[str, Dict[str, List[Union[str, Callable]]]] = {}
for data_table_name, column in self._walk_columns():
data_table_column_name = column.data_table_name
for value_translation_model in column.value_translations:
value_translation_str = value_translation_model.value
value_translation_type = value_translation_model.type
if data_table_name not in by_column:
by_column[data_table_name] = {}
if data_table_column_name not in by_column[data_table_name]:
by_column[data_table_name][data_table_column_name] = []
value_translation: Union[str, Callable]
if value_translation_type == "function":
if value_translation_str in VALUE_TRANSLATION_FUNCTIONS:
value_translation = VALUE_TRANSLATION_FUNCTIONS[value_translation_str]
else:
raise ValueError(f"Unsupported value translation function: '{value_translation}'")
else:
assert value_translation_type == DEFAULT_VALUE_TRANSLATION_TYPE, ValueError(
f"Unsupported value translation type: '{value_translation_type}'"
)
value_translation = value_translation_str
by_column[data_table_name][data_table_column_name].append(value_translation)
return by_column
class RepoInfo(BaseModel):
tool_shed: str
name: str
owner: str
installed_changeset_revision: str
class Config:
# we use dictionary equality (yuk) so definitely make sure this is okay.
extra = Extra.forbid
class DataTableBundle(BaseModel):
processor_description: DataTableBundleProcessorDescription
data_tables: dict
output_name: Optional[str] = None
repo_info: Optional[RepoInfo] = None
def _xml_to_data_table_output_column_move(move_elem: Element) -> DataTableBundleProcessorDataTableOutputColumnMove:
move_type = move_elem.get("type", "directory")
relativize_symlinks = move_elem.get(
"relativize_symlinks", False
) # TODO: should we instead always relativize links?
source_elem = move_elem.find("source")
if source_elem is None:
source_base = None
source_value = ""
else:
source_base = source_elem.get("base", None)
source_value = source_elem.text
target_elem = move_elem.find("target")
if target_elem is None:
target_base = None
target_value = ""
else:
target_base = target_elem.get("base", None)
target_value = target_elem.text
return DataTableBundleProcessorDataTableOutputColumnMove(
type=move_type,
source_base=source_base,
source_value=source_value,
target_base=target_base,
target_value=target_value,
relativize_symlinks=relativize_symlinks,
)
def _xml_to_data_table_output_column_translation(
value_translation_elem: Element,
) -> Optional[DataTableBundleProcessorDataTableOutputColumnTranslation]:
value_translation = value_translation_elem.text
if value_translation is not None:
value_translation_type = value_translation_elem.get("type", DEFAULT_VALUE_TRANSLATION_TYPE)
return DataTableBundleProcessorDataTableOutputColumnTranslation(
value=value_translation, type=value_translation_type
)
else:
return None
def _xml_to_data_table_output_column(column_elem: Element) -> DataTableBundleProcessorDataTableOutputColumn:
column_name = column_elem.get("name", None)
assert column_name is not None, "Name is required for column entry"
data_table_column_name = column_elem.get("data_table_name", column_name)
output_ref = column_elem.get("output_ref", None)
value_translation_elems = column_elem.findall("value_translation")
value_translations = []
if value_translation_elems is not None:
for value_translation_elem in value_translation_elems:
value_translation = _xml_to_data_table_output_column_translation(value_translation_elem)
if value_translation is None:
continue
value_translations.append(value_translation)
moves = []
for move_elem in column_elem.findall("move"):
moves.append(_xml_to_data_table_output_column_move(move_elem))
return DataTableBundleProcessorDataTableOutputColumn(
name=column_name,
data_table_name=data_table_column_name,
output_ref=output_ref,
value_translations=value_translations,
moves=moves,
)
def _xml_to_data_table_output(output_elem: Optional[Element]) -> Optional[DataTableBundleProcessorDataTableOutput]:
if output_elem is not None:
columns = []
for column_elem in output_elem.findall("column"):
columns.append(_xml_to_data_table_output_column(column_elem))
return DataTableBundleProcessorDataTableOutput(columns=columns)
else:
return None
def _xml_to_data_table(data_table_elem: Element) -> DataTableBundleProcessorDataTable:
data_table_name = data_table_elem.get("name")
assert data_table_name is not None, "A name is required for a data table entry"
output_elem = data_table_elem.find("output")
output = _xml_to_data_table_output(output_elem)
return DataTableBundleProcessorDataTable(name=data_table_name, output=output)
def convert_data_tables_xml(elem: Element) -> DataTableBundleProcessorDescription:
undeclared_tables = asbool(elem.get("undeclared_tables", False))
data_tables = []
for data_table_elem in elem.findall("data_table"):
data_tables.append(_xml_to_data_table(data_table_elem))
return DataTableBundleProcessorDescription(undeclared_tables=undeclared_tables, data_tables=data_tables)
lib/galaxy/tools/__init__.py
+9 -1
View File
@@ -3056,7 +3056,15 @@ class DataManagerTool(OutputParameterJSONTool):
assert (
data_manager is not None
), f"Invalid data manager ({data_manager_id}) requested. It may have been removed before the job completed."
data_manager.process_result(out_data)
data_manager_mode = param_dict.get("__data_manager_mode", "populate")
if data_manager_mode == "populate":
data_manager.process_result(out_data)
elif data_manager_mode == "dry_run":
pass
elif data_manager_mode == "bundle":
data_manager.write_bundle(out_data)
else:
raise Exception("Unknown data manager mode encountered type...")
def get_default_history_by_trans(self, trans, create=False):
def _create_data_manager_history(user):
lib/galaxy/tools/data_manager/manager.py
+44 -269
View File
@@ -1,35 +1,27 @@
import errno
import json
import logging
import os
from typing import (
Any,
cast,
Dict,
List,
Optional,
)
from pydantic import BaseModel
from typing_extensions import (
Literal,
Protocol,
TypedDict,
)
from typing_extensions import Protocol
from galaxy import util
from galaxy.structured_app import StructuredApp
from galaxy.tools.data import TabularToolDataTable
from galaxy.tools.data.bundles.models import convert_data_tables_xml
from galaxy.tool_util.data import (
BundleProcessingOptions,
OutputDataset,
)
from galaxy.tool_util.data.bundles.models import (
convert_data_tables_xml,
RepoInfo,
)
from galaxy.util import Element
from galaxy.util.template import fill_template
log = logging.getLogger(__name__)
SUPPORTED_DATA_TABLE_TYPES = TabularToolDataTable
VALUE_TRANSLATION_FUNCTIONS = dict(abspath=os.path.abspath)
DEFAULT_VALUE_TRANSLATION_TYPE = "template"
class DataManagers:
data_managers: Dict[str, "DataManager"]
@@ -102,7 +94,7 @@ class DataManagers:
if data_manager.id in self.data_managers:
log.warning(f"A data manager has been defined twice: {data_manager.id} ")
self.data_managers[data_manager.id] = data_manager
for data_table_name in data_manager.data_tables.keys():
for data_table_name in data_manager.data_table_names:
if data_table_name not in self.managed_data_tables:
self.managed_data_tables[data_table_name] = []
self.managed_data_tables[data_table_name].append(data_manager)
@@ -121,44 +113,22 @@ class DataManagers:
if data_manager.tool:
self.app.toolbox.remove_tool_by_id(data_manager.tool.id)
# determine if any data_tables are no longer tracked
for data_table_name in data_manager.data_tables.keys():
for data_table_name in data_manager.data_table_names:
remove_data_table_tracking = True
for other_data_manager in self.data_managers.values():
if data_table_name in other_data_manager.data_tables:
if data_table_name in other_data_manager.data_table_names:
remove_data_table_tracking = False
break
if remove_data_table_tracking and data_table_name in self.managed_data_tables:
del self.managed_data_tables[data_table_name]
class OutputDataset(Protocol):
file_name: str
extra_files_path: str
def extra_files_path_exists(self) -> bool:
...
class Tool(Protocol):
name: str
description: str
version: str
MoveDict = TypedDict(
"MoveDict",
{
"type": str,
"source_base": Optional[str],
"source_value": str,
"target_base": Optional[str],
"target_value": Optional[str],
"relativize_symlinks": bool,
},
)
RepoInfo = TypedDict("RepoInfo", {"tool_shed": str, "name": str, "owner": str, "installed_changeset_revision": str})
class DataManager:
GUID_TYPE = "data_manager"
DEFAULT_VERSION = "0.0.1"
@@ -173,11 +143,7 @@ class DataManager:
self.version = self.DEFAULT_VERSION
self.guid = None
self.tool = None
self.data_tables: Dict[str, Dict[str, Any]] = {}
self.output_ref_by_data_table: Dict[str, Dict[str, str]] = {}
self.move_by_data_table_column: Dict[str, Dict[str, MoveDict]] = {}
self.value_translation_by_data_table_column: Dict[str, Dict[str, List[Literal["abspath"]]]] = {}
self.tool_shed_repository_info_dict: Optional[RepoInfo] = None
self.tool_shed_repository_info: Optional[RepoInfo] = None
self.undeclared_tables = False
if elem is not None:
self._load_from_element(elem, tool_path or self.data_managers.tool_path)
@@ -201,7 +167,7 @@ class DataManager:
tool_guid = tool_elem.get("guid")
# need to determine repository info so that dependencies will work correctly
tool_shed_repository = self.data_managers.app.toolbox.get_tool_repository_from_xml_item(tool_elem, path)
self.tool_shed_repository_info_dict = dict(
self.tool_shed_repository_info = RepoInfo(
tool_shed=tool_shed_repository.tool_shed,
name=tool_shed_repository.name,
owner=tool_shed_repository.owner,
@@ -225,52 +191,16 @@ class DataManager:
self.name = elem.get("name", self.tool.name)
self.description = elem.get("description", self.tool.description)
self.version = elem.get("version", self.tool.version)
processor_description = convert_data_tables_xml(elem)
self.undeclared_tables = processor_description.undeclared_tables
for data_table in processor_description.data_tables:
data_table_name = data_table.name
if data_table_name not in self.data_tables:
self.data_tables[data_table_name] = {}
output = data_table.output
if output:
for column in output.columns:
column_name = column.name
data_table_column_name = column.data_table_name
self.data_tables[data_table_name][data_table_column_name] = column_name
output_ref = column.output_ref
if output_ref is not None:
if data_table_name not in self.output_ref_by_data_table:
self.output_ref_by_data_table[data_table_name] = {}
self.output_ref_by_data_table[data_table_name][data_table_column_name] = output_ref
for value_translation_model in column.value_translations:
value_translation = value_translation_model.value
value_translation_type = value_translation.type
if data_table_name not in self.value_translation_by_data_table_column:
self.value_translation_by_data_table_column[data_table_name] = {}
if data_table_column_name not in self.value_translation_by_data_table_column[data_table_name]:
self.value_translation_by_data_table_column[data_table_name][data_table_column_name] = []
if value_translation_type == "function":
if value_translation in VALUE_TRANSLATION_FUNCTIONS:
value_translation = VALUE_TRANSLATION_FUNCTIONS[value_translation]
else:
raise ValueError(f"Unsupported value translation function: '{value_translation}'")
else:
assert value_translation_type == DEFAULT_VALUE_TRANSLATION_TYPE, ValueError(
f"Unsupported value translation type: '{value_translation_type}'"
)
self.value_translation_by_data_table_column[data_table_name][data_table_column_name].append(
value_translation
)
for move in column.moves:
if data_table_name not in self.move_by_data_table_column:
self.move_by_data_table_column[data_table_name] = {}
self.move_by_data_table_column[data_table_name][data_table_column_name] = move.dict()
self.processor_description = convert_data_tables_xml(elem)
@property
def id(self):
return self.guid or self.declared_id # if we have a guid, we will use that as the data_manager id
@property
def data_table_names(self):
return self.processor_description.data_table_names
def _load_tool(
self, tool_filename, guid=None, data_manager_id=None, tool_shed_repository_id=None, tool_shed_repository=None
):
@@ -289,184 +219,30 @@ class DataManager:
def process_result(self, out_data: Dict[str, OutputDataset]) -> None:
tool_data_tables = self.data_managers.app.tool_data_tables
data_manager_dicts: Dict[str, Any] = {}
data_manager_dict: Dict[str, Any] = {}
# TODO: fix this merging below
for output_name, output_dataset in out_data.items():
try:
output_dict = json.loads(open(output_dataset.file_name).read())
except Exception as e:
log.warning(f'Error reading DataManagerTool json for "{output_name}": {e}')
continue
data_manager_dicts[output_name] = output_dict
for key, value in output_dict.items():
if key not in data_manager_dict:
data_manager_dict[key] = {}
data_manager_dict[key].update(value)
data_manager_dict.update(output_dict)
data_tables_dict = data_manager_dict.get("data_tables", {})
for data_table_name in self.data_tables.keys():
data_table_values = data_tables_dict.pop(data_table_name, None)
if not data_table_values:
log.warning(
f'No values for data table "{data_table_name}" were returned by the data manager "{self.id}".'
)
continue # next data table
data_table_remove_values = None
if isinstance(data_table_values, dict):
values_to_add = data_table_values.get("add")
data_table_remove_values = data_table_values.get("remove")
if values_to_add or data_table_remove_values:
# We don't have an old style data table definition
data_table_values = values_to_add
data_table = tool_data_tables.get(data_table_name, None)
if data_table is None:
log.error(
f'The data manager "{self.id}" returned an unknown data table "{data_table_name}" with new entries "{data_table_values}". These entries will not be created. Please confirm that an entry for "{data_table_name}" exists in your "tool_data_table_conf.xml" file.'
)
continue # next table name
if not isinstance(data_table, SUPPORTED_DATA_TABLE_TYPES):
log.error(
f'The data manager "{self.id}" returned an unsupported data table "{data_table_name}" with type "{type(data_table)}" with new entries "{data_table_values}". These entries will not be created. Please confirm that the data table is of a supported type ({SUPPORTED_DATA_TABLE_TYPES}).'
)
continue # next table name
output_ref_values = {}
if data_table_name in self.output_ref_by_data_table:
for data_table_column, output_ref in self.output_ref_by_data_table[data_table_name].items():
output_ref_dataset = out_data.get(output_ref, None)
assert output_ref_dataset is not None, "Referenced output was not found."
output_ref_values[data_table_column] = output_ref_dataset
if not isinstance(data_table_values, list):
data_table_values = [data_table_values] if data_table_values else []
if not isinstance(data_table_remove_values, list):
data_table_remove_values = [data_table_remove_values] if data_table_remove_values else []
for data_table_row in data_table_values:
data_table_value = dict(**data_table_row) # keep original values here
for (
name
) in (
data_table_row.keys()
): # FIXME: need to loop through here based upon order listed in data_manager config
if name in output_ref_values:
self.process_move(
data_table_name, name, output_ref_values[name].extra_files_path, **data_table_value
)
data_table_value[name] = self.process_value_translation(
data_table_name, name, **data_table_value
)
data_table.add_entry(data_table_value, persist=True, entry_source=self)
# Removes data table entries
for data_table_row in data_table_remove_values:
data_table_value = dict(**data_table_row) # keep original values here
data_table.remove_entry(list(data_table_value.values()))
options = BundleProcessingOptions(
what=f"data manager '{self.id}'",
data_manager_path=self._data_manager_path,
target_config_file=self.data_managers.filename,
)
updated_data_tables = tool_data_tables.process_bundle(
out_data,
self.processor_description,
self._repo_info,
options,
)
for data_table_name in updated_data_tables:
self._reload(data_table_name)
if self.undeclared_tables and data_tables_dict:
# We handle the data move, by just moving all the data out of the extra files path
# moving a directory and the target already exists, we move the contents instead
log.debug("Attempting to add entries for undeclared tables: %s.", ", ".join(data_tables_dict.keys()))
for ref_file in out_data.values():
if ref_file.extra_files_path_exists():
util.move_merge(ref_file.extra_files_path, self._data_manager_path)
path_column_names = ["path"]
for data_table_name, data_table_values in data_tables_dict.items():
data_table = tool_data_tables.get(data_table_name, None)
if not isinstance(data_table_values, list):
data_table_values = [data_table_values]
for data_table_row in data_table_values:
data_table_value = dict(**data_table_row) # keep original values here
for name, value in data_table_row.items():
if name in path_column_names:
data_table_value[name] = os.path.abspath(os.path.join(self._data_manager_path, value))
data_table.add_entry(data_table_value, persist=True, entry_source=self)
self._reload(data_table_name)
else:
for data_table_name, data_table_values in data_tables_dict.items():
# tool returned extra data table entries, but data table was not declared in data manager
# do not add these values, but do provide messages
log.warning(
f'The data manager "{self.id}" returned an undeclared data table "{data_table_name}" with new entries "{data_table_values}". These entries will not be created. Please confirm that an entry for "{data_table_name}" exists in your "{self.data_managers.filename}" file.'
)
def process_move(self, data_table_name, column_name, source_base_path, relative_symlinks=False, **kwd):
if (
data_table_name in self.move_by_data_table_column
and column_name in self.move_by_data_table_column[data_table_name]
):
move_dict = self.move_by_data_table_column[data_table_name][column_name]
source = move_dict["source_base"]
if source is None:
source = source_base_path
else:
source = fill_template(
source,
GALAXY_DATA_MANAGER_DATA_PATH=self._data_manager_path,
**kwd,
).strip()
if move_dict["source_value"]:
source = os.path.join(
source,
fill_template(
move_dict["source_value"],
GALAXY_DATA_MANAGER_DATA_PATH=self._data_manager_path,
**kwd,
).strip(),
)
target = move_dict["target_base"]
if target is None:
target = self._data_manager_path
else:
target = fill_template(
target,
GALAXY_DATA_MANAGER_DATA_PATH=self._data_manager_path,
**kwd,
).strip()
if move_dict["target_value"]:
target = os.path.join(
target,
fill_template(
move_dict["target_value"],
GALAXY_DATA_MANAGER_DATA_PATH=self._data_manager_path,
**kwd,
).strip(),
)
if move_dict["type"] == "file":
dirs = os.path.split(target)[0]
try:
os.makedirs(dirs)
except OSError as e:
if e.errno != errno.EEXIST:
raise e
# moving a directory and the target already exists, we move the contents instead
if os.path.exists(source):
util.move_merge(source, target)
if move_dict.get("relativize_symlinks", False):
util.relativize_symlinks(target)
return True
return False
def process_value_translation(self, data_table_name, column_name, **kwd):
value = kwd.get(column_name)
if (
data_table_name in self.value_translation_by_data_table_column
and column_name in self.value_translation_by_data_table_column[data_table_name]
):
for value_translation in self.value_translation_by_data_table_column[data_table_name][column_name]:
if isinstance(value_translation, str):
value = fill_template(
value_translation,
GALAXY_DATA_MANAGER_DATA_PATH=self._data_manager_path,
**kwd,
).strip()
else:
value = value_translation(value)
return value
def write_bundle(
self,
out_data: Dict[str, OutputDataset],
) -> None:
tool_data_tables = self.data_managers.app.tool_data_tables
tool_data_tables.write_bundle(
out_data,
self.processor_description,
self._repo_info,
)
@property
def _data_manager_path(self) -> str:
@@ -477,7 +253,6 @@ class DataManager:
"reload_tool_data_tables", noop_self=True, kwargs={"table_name": data_table_name}
)
def get_tool_shed_repository_info_dict(self) -> Optional[dict]:
return (
cast(dict, self.tool_shed_repository_info_dict) if self.tool_shed_repository_info_dict is not None else None
)
@property
def _repo_info(self) -> Optional[RepoInfo]:
return self.tool_shed_repository_info
lib/galaxy/tools/execute.py
+2
View File
@@ -95,6 +95,8 @@ def execute(
"internals.galaxy.tools.execute.job_single", SINGLE_EXECUTION_SUCCESS_MESSAGE
)
params = execution_slice.param_combination
if "__data_manager_mode" in mapping_params.param_template:
params["__data_manager_mode"] = mapping_params.param_template["__data_manager_mode"]
if workflow_invocation_uuid:
params["__workflow_invocation_uuid__"] = workflow_invocation_uuid
elif "__workflow_invocation_uuid__" in params:
lib/galaxy/util/compression_utils.py
+14
View File
@@ -4,6 +4,7 @@ import io
import logging
import os
import tarfile
import tempfile
import zipfile
from typing import (
Any,
@@ -152,6 +153,19 @@ def file_iter(fname: str, sep: Optional[Any] = None) -> Generator[Union[List[byt
ArchiveMemberType = Union[tarfile.TarInfo, zipfile.ZipInfo]
def decompress_bytes_to_directory(content: bytes) -> str:
temp_directory = tempfile.mkdtemp()
with tempfile.NamedTemporaryFile(delete=False) as fp:
fp.write(content)
fp.close()
return CompressedFile(fp.name).extract(temp_directory)
def decompress_path_to_directory(path: str) -> str:
temp_directory = tempfile.mkdtemp()
return CompressedFile(path).extract(temp_directory)
class CompressedFile:
archive: Union[tarfile.TarFile, zipfile.ZipFile]
lib/galaxy/webapps/galaxy/api/tool_data.py
+29 -1
View File
@@ -1,6 +1,18 @@
from fastapi import Path
from fastapi import (
Body,
Path,
)
from pydantic import (
BaseModel,
Field,
)
from galaxy.celery.tasks import import_data_bundle
from galaxy.managers.tool_data import ToolDataManager
from galaxy.schema.schema import (
AsyncTaskResultSummary,
ImportToolDataBundleSource,
)
from galaxy.tool_util.data._schema import (
ToolDataDetails,
ToolDataEntryList,
@@ -8,6 +20,7 @@ from galaxy.tool_util.data._schema import (
ToolDataItem,
)
from galaxy.webapps.base.api import GalaxyFileResponse
from galaxy.webapps.galaxy.services.base import async_task_summary
from . import (
depends,
Router,
@@ -29,6 +42,10 @@ ToolDataTableFieldName = Path(
)
class ImportToolDataBundle(BaseModel):
source: ImportToolDataBundleSource = Field(..., discriminator="src")
@router.cbv
class FastAPIToolData:
tool_data_manager: ToolDataManager = depends(ToolDataManager)
@@ -43,6 +60,17 @@ class FastAPIToolData:
"""Get the list of all available data tables."""
return self.tool_data_manager.index()
@router.post(
"/api/tool_data",
summary="Import a data manager bundle",
require_admin=True,
)
async def create(self, import_bundle_model: ImportToolDataBundle = Body(...)) -> AsyncTaskResultSummary:
source = import_bundle_model.source
result = import_data_bundle.delay(**source.dict())
summary = async_task_summary(result)
return summary
@router.get(
"/api/tool_data/{table_name}",
summary="Get details of a given data table",
lib/galaxy/webapps/galaxy/services/tools.py
+2 -1
View File
@@ -163,7 +163,8 @@ class ToolsService(ServiceBase):
)
input_format = str(payload.get("input_format", "legacy"))
if "data_manager_mode" in payload:
incoming["__data_manager_mode"] = payload["data_manager_mode"]
vars = tool.handle_input(
trans, incoming, history=target_history, use_cached_job=use_cached_job, input_format=input_format
)
lib/galaxy_test/base/populators.py
+5 -1
View File
@@ -901,13 +901,17 @@ class BaseDatasetPopulator(BasePopulator):
assert "id" in create_response_json
return create_response_json
def get_history_dataset_content(self, history_id: str, wait=True, filename=None, type="text", raw=False, **kwds):
def get_history_dataset_content(
self, history_id: str, wait=True, filename=None, type="text", to_ext=None, raw=False, **kwds
):
dataset_id = self.__history_content_id(history_id, wait=wait, **kwds)
data = {}
if filename:
data["filename"] = filename
if raw:
data["raw"] = True
if to_ext is not None:
data["to_ext"] = to_ext
display_response = self._get_contents_request(history_id, f"/{dataset_id}/display", data=data)
assert display_response.status_code == 200, display_response.text
if type == "text":
test/integration/test_tool_data_bundles.py
+73
View File
@@ -0,0 +1,73 @@
import os
from galaxy.util.compression_utils import decompress_bytes_to_directory
from .test_tool_data_delete import DataManagerIntegrationTestCase
class TestDataBundlesIntegration(DataManagerIntegrationTestCase):
def test_admin_build_data_bundle_by_uri(self):
original_count = self._testbeta_field_count()
history_id = self.dataset_populator.new_history()
payload = self.dataset_populator.run_tool_payload(
tool_id="data_manager",
inputs={"ignored_value": "moo"},
data_manager_mode="bundle",
history_id=history_id,
)
create_response = self._post("tools", data=payload)
create_response.raise_for_status()
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
data_manager_dataset = self.dataset_populator.get_history_dataset_details(history_id)
assert data_manager_dataset["extension"] == "data_manager_json"
post_job_count = self._testbeta_field_count()
assert original_count == post_job_count
content = self.dataset_populator.get_history_dataset_content(
history_id, to_ext="data_manager_json", type="bytes"
)
temp_directory = decompress_bytes_to_directory(content)
uri = f"file://{os.path.normpath(temp_directory)}"
data = {
"source": {
"src": "uri",
"uri": uri,
}
}
task_summary_response = self._post("tool_data", data=data, json=True)
self.dataset_populator.wait_on_task(task_summary_response)
post_import_count = self._testbeta_field_count()
assert original_count + 1 == post_import_count
def test_admin_build_data_bundle_by_dataset(self):
original_count = self._testbeta_field_count()
history_id = self.dataset_populator.new_history()
payload = self.dataset_populator.run_tool_payload(
tool_id="data_manager",
inputs={"ignored_value": "moo"},
data_manager_mode="bundle",
history_id=history_id,
)
create_response = self._post("tools", data=payload)
create_response.raise_for_status()
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
data_manager_dataset = self.dataset_populator.get_history_dataset_details(history_id)
assert data_manager_dataset["extension"] == "data_manager_json"
post_job_count = self._testbeta_field_count()
assert original_count == post_job_count
data = {
"source": {
"src": "hda",
"id": data_manager_dataset["id"],
}
}
task_summary_response = self._post("tool_data", data=data, json=True)
self.dataset_populator.wait_on_task(task_summary_response)
post_import_count = self._testbeta_field_count()
assert original_count + 1 == post_import_count
test/integration/test_tool_data_delete.py
+14 -5
View File
@@ -17,8 +17,10 @@ THIS_DIR = os.path.dirname(__file__)
SOURCE_TOOL_DATA_DIRECTORY = os.path.join(THIS_DIR, os.pardir, "functional", "tool-data")
class TestAdminToolDataIntegration(integration_util.IntegrationTestCase):
class DataManagerIntegrationTestCase(integration_util.IntegrationTestCase):
require_admin_user = True
dataset_populator: DatasetPopulator
def setUp(self):
super().setUp()
@@ -38,10 +40,18 @@ class TestAdminToolDataIntegration(integration_util.IntegrationTestCase):
config["tool_data_path"] = cls.temp_tool_data_dir
config["tool_data_table_config_path"] = cls.temp_tool_data_tables_file
def test_admin_delete_data_table_entry(self):
def _testbase_fields(self):
show_response = self._get("tool_data/testbeta")
original_count = len(show_response.json()["fields"])
show_response.raise_for_status()
return show_response.json()["fields"]
def _testbeta_field_count(self) -> int:
return len(self._testbase_fields())
class TestAdminToolDataIntegration(DataManagerIntegrationTestCase):
def test_admin_delete_data_table_entry(self):
original_count = self._testbeta_field_count()
history_id = self.dataset_populator.new_history()
payload = self.dataset_populator.run_tool_payload(
tool_id="data_manager",
@@ -52,8 +62,7 @@ class TestAdminToolDataIntegration(integration_util.IntegrationTestCase):
create_response.raise_for_status()
self.dataset_populator.wait_for_history(history_id, assert_ok=True)
time.sleep(2)
show_response = self._get("tool_data/testbeta")
updated_fields = show_response.json()["fields"]
updated_fields = self._testbase_fields()
assert len(updated_fields) == original_count + 1
new_field = updated_fields[-1]
url = self._api_url(f"tool_data/testbeta?key={self.galaxy_interactor.api_key}")
test/unit/tool_util/data/__init__.py
View File
test/unit/tool_util/test_tool_data.py → test/unit/tool_util/data/conftest.py
-64
View File
@@ -8,13 +8,6 @@ data2 data2name ${__HERE__}/data2/entry.txt
"""
LOC_ALPHA_CONTENTS_V2 = """
data1 data1name ${__HERE__}/data1/entry.txt
data2 data2name ${__HERE__}/data2/entry.txt
data3 data3name ${__HERE__}/data3/entry.txt
"""
LOC_BETA_CONTENTS_1 = """
beta1 beta1name ${__HERE__}/beta1/entry.txt
"""
@@ -104,60 +97,3 @@ def _write_loc_files(tmp_path):
beta2.mkdir()
beta2_entry = beta2 / "entry.txt"
beta2_entry.write_text("This is beta 2.")
def test_data_tables_as_dictionary(tdt_manager):
assert "testalpha" in tdt_manager.data_tables
assert "testdelta" not in tdt_manager.data_tables
def test_to_dict(tdt_manager):
as_dict = tdt_manager.to_dict()
assert "testalpha" in as_dict
assert "testdelta" not in as_dict
testalpha_as_dict = as_dict["testalpha"]
assert "columns" in testalpha_as_dict
def test_index(tdt_manager):
index = tdt_manager.index()
assert len(index.__root__) >= 1
entry = index.find_entry("testalpha")
assert entry
entry = index.find_entry("testomega")
assert not entry
def test_reload(tdt_manager, tmp_path):
assert len(tdt_manager["testalpha"].data) == 2
loc1 = tmp_path / "testalpha.loc"
loc1.write_text(LOC_ALPHA_CONTENTS_V2)
tdt_manager.reload_tables()
assert len(tdt_manager["testalpha"].data) == 3
def test_reload_by_path(tdt_manager, tmp_path):
assert len(tdt_manager["testalpha"].data) == 2
loc1 = tmp_path / "testalpha.loc"
loc1.write_text(LOC_ALPHA_CONTENTS_V2)
tdt_manager.reload_tables(path=str(loc1))
assert len(tdt_manager["testalpha"].data) == 3
def test_reload_by_name(tdt_manager, tmp_path):
assert len(tdt_manager["testalpha"].data) == 2
loc1 = tmp_path / "testalpha.loc"
loc1.write_text(LOC_ALPHA_CONTENTS_V2)
tdt_manager.reload_tables("testalpha")
assert len(tdt_manager["testalpha"].data) == 3
def test_merging_tables(merged_tdt_manager):
assert len(merged_tdt_manager["testbeta"].data) == 2
def test_to_json(merged_tdt_manager, tmp_path):
json_path = tmp_path / "as_json.json"
assert not json_path.exists()
merged_tdt_manager.to_json(json_path)
assert json_path.exists()
test/unit/tool_util/example_data_managers/manual.xml → test/unit/tool_util/data/example_data_managers/manual.xml
View File
test/unit/tool_util/example_data_managers/mothur.xml → test/unit/tool_util/data/example_data_managers/mothur.xml
View File
test/unit/tool_util/data/test_tool_data.py
+62
View File
@@ -0,0 +1,62 @@
LOC_ALPHA_CONTENTS_V2 = """
data1 data1name ${__HERE__}/data1/entry.txt
data2 data2name ${__HERE__}/data2/entry.txt
data3 data3name ${__HERE__}/data3/entry.txt
"""
def test_data_tables_as_dictionary(tdt_manager):
assert "testalpha" in tdt_manager.data_tables
assert "testdelta" not in tdt_manager.data_tables
def test_to_dict(tdt_manager):
as_dict = tdt_manager.to_dict()
assert "testalpha" in as_dict
assert "testdelta" not in as_dict
testalpha_as_dict = as_dict["testalpha"]
assert "columns" in testalpha_as_dict
def test_index(tdt_manager):
index = tdt_manager.index()
assert len(index.__root__) >= 1
entry = index.find_entry("testalpha")
assert entry
entry = index.find_entry("testomega")
assert not entry
def test_reload(tdt_manager, tmp_path):
assert len(tdt_manager["testalpha"].data) == 2
loc1 = tmp_path / "testalpha.loc"
loc1.write_text(LOC_ALPHA_CONTENTS_V2)
tdt_manager.reload_tables()
assert len(tdt_manager["testalpha"].data) == 3
def test_reload_by_path(tdt_manager, tmp_path):
assert len(tdt_manager["testalpha"].data) == 2
loc1 = tmp_path / "testalpha.loc"
loc1.write_text(LOC_ALPHA_CONTENTS_V2)
tdt_manager.reload_tables(path=str(loc1))
assert len(tdt_manager["testalpha"].data) == 3
def test_reload_by_name(tdt_manager, tmp_path):
assert len(tdt_manager["testalpha"].data) == 2
loc1 = tmp_path / "testalpha.loc"
loc1.write_text(LOC_ALPHA_CONTENTS_V2)
tdt_manager.reload_tables("testalpha")
assert len(tdt_manager["testalpha"].data) == 3
def test_merging_tables(merged_tdt_manager):
assert len(merged_tdt_manager["testbeta"].data) == 2
def test_to_json(merged_tdt_manager, tmp_path):
json_path = tmp_path / "as_json.json"
assert not json_path.exists()
merged_tdt_manager.to_json(json_path)
assert json_path.exists()
test/unit/tool_util/data/test_tool_data_bundles.py
+248
View File
@@ -0,0 +1,248 @@
import csv
import json
import os
from dataclasses import dataclass
from galaxy.tool_util.data import (
BUNDLE_INDEX_FILE_NAME,
BundleProcessingOptions,
)
from galaxy.tool_util.data.bundles.models import (
convert_data_tables_xml,
DataTableBundleProcessorDescription,
)
from galaxy.util import (
galaxy_directory,
parse_xml,
)
from galaxy.util.resources import resource_path
TOOLS_DIRECTORY = os.path.abspath(os.path.join(galaxy_directory(), "test/functional/tools/"))
def test_xml_parsing() -> None:
path = os.path.join(TOOLS_DIRECTORY, "sample_data_manager_conf.xml")
tree = parse_xml(path)
data_managers_el = tree.getroot()
data_manager_el = data_managers_el.find("data_manager")
description = convert_data_tables_xml(data_manager_el)
assert not description.undeclared_tables
assert len(description.data_tables) == 1
data_table = description.data_tables[0]
output = data_table.output
assert output
columns = output.columns
assert len(columns) == 2
column1 = columns[0]
assert column1.name == "value"
assert column1.output_ref is None
column2 = columns[1]
assert column2.name == "path"
assert column2.output_ref == "out_file"
moves = column2.moves
assert len(moves) == 1
move = moves[0]
assert move.type == "directory"
assert move.relativize_symlinks is True
assert move.target_base == "${GALAXY_DATA_MANAGER_DATA_PATH}"
assert move.target_value == "testbeta/${value}"
assert move.source_base is None
assert move.source_value == ""
def test_parsing_manual() -> None:
path = resource_path(__package__, "example_data_managers/manual.xml")
tree = parse_xml(path)
data_managers_el = tree.getroot()
data_manager_el = data_managers_el.find("data_manager")
description = convert_data_tables_xml(data_manager_el)
assert description.undeclared_tables
assert len(description.data_tables) == 0
def test_parsing_mothur() -> None:
path = resource_path(__package__, "example_data_managers/mothur.xml")
tree = parse_xml(path)
data_managers_el = tree.getroot()
data_manager_el = data_managers_el.find("data_manager")
description = convert_data_tables_xml(data_manager_el)
assert not description.undeclared_tables
assert len(description.data_tables) == 4
@dataclass
class OutputDataset:
file_name: str
extra_files_path: str
ext: str = "data_manager_json"
def extra_files_path_exists(self) -> bool:
return os.path.exists(self.extra_files_path)
def prepare_typical_output_and_description(tmp_path):
target_path = tmp_path / "newvalue.txt"
target_path.write_text("Moo Cow")
output = {"data_tables": {"testalpha": [{"value": "newvalue", "name": "mynewname", "path": "newvalue.txt"}]}}
output_dataset_path = tmp_path / "output.dat"
output_dataset_path.write_text(json.dumps(output))
extra_files_path = tmp_path / "extra"
extra_files_path.mkdir()
output_dataset = OutputDataset(
output_dataset_path,
extra_files_path,
)
out_data = {"out1": output_dataset}
data_table = {
"name": "testalpha",
"output": {
"columns": [
{
"name": "value",
},
{
"name": "name",
},
{
"name": "path",
"data_table_name": "path",
"output_ref": "out1",
"moves": [
{
"type": "directory",
"relativize_symlinks": True,
"target_value": "testalpha/${value}",
"target_base": "${GALAXY_DATA_MANAGER_DATA_PATH}",
}
],
"value_translations": [
{"value": "${GALAXY_DATA_MANAGER_DATA_PATH}/testalpha/${value}/${path}", "type": "template"},
{"value": "abspath", "type": "function"},
],
},
]
},
}
process_description = DataTableBundleProcessorDescription(
**{
"undeclared_tables": False,
"data_tables": [data_table],
}
)
return out_data, process_description
def test_typical_processing(tdt_manager, tmp_path):
options = BundleProcessingOptions(
what="data manager 'mock'",
data_manager_path=str(tmp_path),
target_config_file=str(tmp_path / "sample_data_managers_conf.xml"),
)
out_data, process_description = prepare_typical_output_and_description(tmp_path)
repo_info = None
tdt_manager.process_bundle(
out_data,
process_description,
repo_info,
options,
)
loc1 = tmp_path / "testalpha.loc"
new_row = _last_row(loc1)
assert new_row[0] == "newvalue"
assert new_row[1] == "mynewname"
assert new_row[2] == str(tmp_path / "testalpha" / "newvalue" / "newvalue.txt")
def test_write_bundle(tdt_manager, tmp_path):
out_data, process_description = prepare_typical_output_and_description(tmp_path)
tdt_manager.write_bundle(
out_data,
process_description,
repo_info=None,
)
extra = tmp_path / "extra"
bundle_index_json_path = extra / BUNDLE_INDEX_FILE_NAME
assert bundle_index_json_path.exists()
with open(bundle_index_json_path, "r") as f:
bundle_index = json.load(f)
assert "processor_description" in bundle_index
assert "data_tables" in bundle_index
assert "output_name" in bundle_index
def test_import_bundle(tdt_manager, tmp_path):
out_data, process_description = prepare_typical_output_and_description(tmp_path)
tdt_manager.write_bundle(
out_data,
process_description,
None,
)
# Writing the bundle didn't update the loc files.
loc1 = tmp_path / "testalpha.loc"
new_row = _last_row(loc1)
assert new_row[0] != "newvalue"
options = BundleProcessingOptions(
what="data manager 'mock'",
data_manager_path=str(tmp_path),
target_config_file=str(tmp_path / "sample_data_managers_conf.xml"),
)
tdt_manager.import_bundle(
str(tmp_path / "extra"),
options,
)
# Importing the bundle does update the loc files though.
loc1 = tmp_path / "testalpha.loc"
new_row = _last_row(loc1)
assert new_row[0] == "newvalue"
assert new_row[1] == "mynewname"
assert new_row[2] == str(tmp_path / "testalpha" / "newvalue" / "newvalue.txt")
def test_undeclared_tables(tdt_manager, tmp_path):
options = BundleProcessingOptions(
what="data manager 'mock'",
data_manager_path=str(tmp_path),
target_config_file=str(tmp_path / "sample_data_managers_conf.xml"),
)
target_path = tmp_path / "newvalue.txt"
target_path.write_text("Moo Cow")
output = {"data_tables": {"testalpha": [{"value": "newvalue", "name": "mynewname", "path": "newvalue.txt"}]}}
output_dataset_path = tmp_path / "output.dat"
import json
output_dataset_path.write_text(json.dumps(output))
extra_files_path = tmp_path / "extra"
output_dataset = OutputDataset(
output_dataset_path,
extra_files_path,
)
out_data = {"out1": output_dataset}
process_description = DataTableBundleProcessorDescription(
**{
"undeclared_tables": True,
"data_tables": [],
}
)
repo_info = None
tdt_manager.process_bundle(
out_data,
process_description,
repo_info,
options,
)
loc1 = tmp_path / "testalpha.loc"
new_row = _last_row(loc1)
assert new_row[0] == "newvalue"
assert new_row[1] == "mynewname"
assert new_row[2] == str(tmp_path / "newvalue.txt")
def _last_row(loc_file):
with open(loc_file, "r") as file:
rows = csv.reader(file, delimiter="\t")
for row in rows:
last_row = row
return last_row
test/unit/tool_util/test_tool_data_bundles.py
-60
View File
@@ -1,60 +0,0 @@
import os
from galaxy.tool_util.data.bundles.models import convert_data_tables_xml
from galaxy.util import (
galaxy_directory,
parse_xml,
)
from galaxy.util.resources import resource_path
TOOLS_DIRECTORY = os.path.abspath(os.path.join(galaxy_directory(), "test/functional/tools/"))
def test_xml_parsing() -> None:
path = os.path.join(TOOLS_DIRECTORY, "sample_data_manager_conf.xml")
tree = parse_xml(path)
data_managers_el = tree.getroot()
data_manager_el = data_managers_el.find("data_manager")
description = convert_data_tables_xml(data_manager_el)
assert not description.undeclared_tables
assert len(description.data_tables) == 1
data_table = description.data_tables[0]
output = data_table.output
columns = output.columns
assert len(columns) == 2
column1 = columns[0]
assert column1.name == "value"
assert column1.output_ref is None
column2 = columns[1]
assert column2.name == "path"
assert column2.output_ref == "out_file"
moves = column2.moves
assert len(moves) == 1
move = moves[0]
assert move.type == "directory"
assert move.relativize_symlinks is True
assert move.target_base == "${GALAXY_DATA_MANAGER_DATA_PATH}"
assert move.target_value == "testbeta/${value}"
assert move.source_base is None
assert move.source_value == ""
def test_parsing_manual() -> None:
path = resource_path(__package__, "example_data_managers/manual.xml")
tree = parse_xml(path)
data_managers_el = tree.getroot()
data_manager_el = data_managers_el.find("data_manager")
description = convert_data_tables_xml(data_manager_el)
assert description.undeclared_tables
assert len(description.data_tables) == 0
def test_parsing_mothur() -> None:
path = resource_path(__package__, "example_data_managers/mothur.xml")
tree = parse_xml(path)
data_managers_el = tree.getroot()
data_manager_el = data_managers_el.find("data_manager")
description = convert_data_tables_xml(data_manager_el)
assert not description.undeclared_tables
assert len(description.data_tables) == 4