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

Merge branch 'release_21.09' into dev

Browse Source
This commit is contained in:
mvdbeek
2021-10-20 17:24:41 +02:00
parent 5e8531015c 0e25970adf
commit 0e11349685
29 changed files with 220 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/api.yaml
+1
View File
@@ -2,6 +2,7 @@ name: API tests
on: [push, pull_request]
env:
GALAXY_TEST_DBURI: 'postgresql://postgres:postgres@localhost:5432/galaxy?client_encoding=utf8'
GALAXY_TEST_RAISE_EXCEPTION_ON_HISTORYLESS_HDA: '1'
concurrency:
group: api-${{ github.ref }}
cancel-in-progress: true
.github/workflows/api_paste.yaml
+1
View File
@@ -3,6 +3,7 @@ on: [push, pull_request]
env:
GALAXY_TEST_DBURI: 'postgresql://postgres:postgres@localhost:5432/galaxy?client_encoding=utf8'
GALAXY_TEST_USE_UVICORN: false
GALAXY_TEST_RAISE_EXCEPTION_ON_HISTORYLESS_HDA: '1'
concurrency:
group: api-legacy-${{ github.ref }}
cancel-in-progress: true
.github/workflows/converter_tests.yaml
+3 -1
View File
@@ -1,5 +1,7 @@
name: Converter tests
on: [push, pull_request]
env:
GALAXY_TEST_RAISE_EXCEPTION_ON_HISTORYLESS_HDA: '1'
concurrency:
group: converter-${{ github.ref }}
cancel-in-progress: true
@@ -65,4 +67,4 @@ jobs:
if: failure()
with:
name: Converter test results (${{ matrix.python-version }})
path: tool_test_output.html
path: tool_test_output.html
.github/workflows/framework.yaml
+2
View File
@@ -1,5 +1,7 @@
name: Framework tests
on: [push, pull_request]
env:
GALAXY_TEST_RAISE_EXCEPTION_ON_HISTORYLESS_HDA: '1'
concurrency:
group: framework-${{ github.ref }}
cancel-in-progress: true
.github/workflows/integration.yaml
+1
View File
@@ -6,6 +6,7 @@ concurrency:
env:
GALAXY_TEST_DBURI: 'postgresql://postgres:postgres@localhost:5432/galaxy?client_encoding=utf8'
GALAXY_TEST_AMQP_URL: 'amqp://localhost:5672//'
GALAXY_TEST_RAISE_EXCEPTION_ON_HISTORYLESS_HDA: '1'
jobs:
test:
name: Test
.github/workflows/integration_selenium.yaml
+1
View File
@@ -8,6 +8,7 @@ env:
GALAXY_SKIP_CLIENT_BUILD: '0'
GALAXY_TEST_SELENIUM_RETRIES: 1
YARN_INSTALL_OPTS: --frozen-lockfile
GALAXY_TEST_RAISE_EXCEPTION_ON_HISTORYLESS_HDA: '1'
jobs:
test:
name: Test
.github/workflows/selenium.yaml
+1
View File
@@ -8,6 +8,7 @@ env:
GALAXY_TEST_SKIP_FLAKEY_TESTS_ON_ERROR: 'true'
GALAXY_TEST_SELENIUM_RETRIES: 1
YARN_INSTALL_OPTS: --frozen-lockfile
GALAXY_TEST_RAISE_EXCEPTION_ON_HISTORYLESS_HDA: '1'
jobs:
test:
name: Test
.github/workflows/selenium_beta.yaml
+1
View File
@@ -6,6 +6,7 @@ env:
GALAXY_TEST_SELENIUM_RETRIES: 1
GALAXY_TEST_SELENIUM_BETA_HISTORY: 1
YARN_INSTALL_OPTS: --frozen-lockfile
GALAXY_TEST_RAISE_EXCEPTION_ON_HISTORYLESS_HDA: '1'
jobs:
test:
name: Test
client/src/components/History/providers/HistoryContentProvider/contentPayload.js
+1 -1
View File
@@ -150,7 +150,7 @@ export const contentPayload = (cfg = {}) => {
filter(([a,b]) => !isNaN(a.maxHid) && !isNaN(b.maxHid)),
withLatestFrom(pos$, hid$),
map(([[lastResponse, response], pos, hid]) => {
const updatesAtTop = response.maxHid > lastResponse.maxHid;
const updatesAtTop = response.maxHid >= lastResponse.maxHid;
const scrollerExactlyAtTop = pos.cursor === 0 || pos.key === lastResponse.maxHid;
const fudge = 2;
lib/galaxy/datatypes/test/1.yaml
+7
View File
@@ -0,0 +1,7 @@
test:
- test1
- test2
testing: testing 1
test2:
- Lorem ipsum dolor sit amet, consectetur adipiscing elit. Phasellus ac justo metus. Praesent eget mollis enim, sed facilisis felis. Pellentesque vulputate varius mi, vel eleifend mauris luctus vestibulum. Suspendisse elementum nisi lorem, in consequat nisl commodo nec. Curabitur faucibus nec diam id interdum. Sed lobortis sed libero id feugiat. Duis diam odio, elementum sed suscipit ac, venenatis a eros. Phasellus at sollicitudin ligula. Suspendisse congue, tortor vitae aliquet ornare, diam mi scelerisque eros, vel tempus elit odio ac ligula. Pellentesque ultricies diam ornare tempor sagittis. Donec vel elit ac elit placerat dictum. Pellentesque et diam at sem porttitor blandit.
- Lorem ipsum dolor sit amet, consectetur adipiscing elit. Phasellus ac justo metus. Praesent eget mollis enim, sed facilisis felis. Pellentesque vulputate varius mi, vel eleifend mauris luctus vestibulum. Suspendisse elementum nisi lorem, in consequat nisl commodo nec. Curabitur faucibus nec diam id interdum. Sed lobortis sed libero id feugiat. Duis diam odio, elementum sed suscipit ac, venenatis a eros. Phasellus at sollicitudin ligula. Suspendisse congue, tortor vitae aliquet ornare, diam mi scelerisque eros, vel tempus elit odio ac ligula. Pellentesque ultricies diam ornare tempor sagittis. Donec vel elit ac elit placerat dictum. Pellentesque et diam at sem porttitor blandit.
lib/galaxy/job_execution/output_collect.py
+25 -17
View File
@@ -17,7 +17,6 @@ from galaxy.model.store.discover import (
ModelPersistenceContext,
persist_elements_to_folder,
persist_elements_to_hdca,
persist_extra_files,
persist_hdas,
RegexCollectedDatasetMatch,
SessionlessModelPersistenceContext,
@@ -207,6 +206,13 @@ class JobContext(ModelPersistenceContext, BaseJobContext):
self.object_store = object_store
self.final_job_state = final_job_state
self.flush_per_n_datasets = flush_per_n_datasets
self._tag_handler = None
@property
def tag_handler(self):
if self._tag_handler is None:
self._tag_handler = self.app.tag_handler.create_tag_handler_session()
return self._tag_handler
@property
def work_context(self):
@@ -221,10 +227,6 @@ class JobContext(ModelPersistenceContext, BaseJobContext):
user = None
return user
@property
def tag_handler(self):
return self.app.tag_handler
def persist_object(self, obj):
self.sa_session.add(obj)
@@ -284,7 +286,8 @@ class JobContext(ModelPersistenceContext, BaseJobContext):
def add_datasets_to_history(self, datasets, for_output_dataset=None):
sa_session = self.sa_session
self.job.history.add_datasets(sa_session, datasets)
self.job.history.stage_addition(datasets)
pending_histories = {self.job.history}
if for_output_dataset is not None:
# Need to update all associated output hdas, i.e. history was
# shared with job running
@@ -293,9 +296,11 @@ class JobContext(ModelPersistenceContext, BaseJobContext):
continue
for dataset in datasets:
new_data = dataset.copy()
copied_dataset.history.add_dataset(new_data)
copied_dataset.history.stage_addition(new_data)
pending_histories.add(copied_dataset.history)
sa_session.add(new_data)
sa_session.flush()
for history in pending_histories:
history.add_pending_items()
def output_collection_def(self, name):
tool = self.tool
@@ -388,6 +393,7 @@ def collect_primary_datasets(job_context, output, input_ext):
primary_output_assigned = False
new_outdata_name = None
primary_datasets = {}
storage_callbacks = []
for output_index, (name, outdata) in enumerate(output.items()):
dataset_collectors = [DEFAULT_DATASET_COLLECTOR]
output_def = job_context.output_def(name)
@@ -429,7 +435,11 @@ def collect_primary_datasets(job_context, output, input_ext):
# TODO: should be able to disambiguate files in different directories...
new_primary_filename = os.path.split(filename)[-1]
new_primary_datasets_attributes = job_context.tool_provided_metadata.get_new_dataset_meta_by_basename(name, new_primary_filename)
extra_files = None
if new_primary_datasets_attributes:
extra_files_path = new_primary_datasets_attributes.get('extra_files', None)
if extra_files_path:
extra_files = os.path.join(job_working_directory, extra_files_path)
primary_data = job_context.create_dataset(
ext,
designation,
@@ -437,19 +447,15 @@ def collect_primary_datasets(job_context, output, input_ext):
dbkey,
new_primary_name,
filename,
extra_files=extra_files,
info=info,
init_from=outdata,
dataset_attributes=new_primary_datasets_attributes,
creating_job_id=job_context.get_job_id() if job_context else None
creating_job_id=job_context.get_job_id() if job_context else None,
storage_callbacks=storage_callbacks
)
# Associate new dataset with job
job_context.add_output_dataset_association(f'__new_primary_file_{name}|{designation}__', primary_data)
if new_primary_datasets_attributes:
extra_files_path = new_primary_datasets_attributes.get('extra_files', None)
if extra_files_path:
extra_files_path_joined = os.path.join(job_working_directory, extra_files_path)
persist_extra_files(job_context.object_store, extra_files_path_joined, primary_data)
job_context.add_datasets_to_history([primary_data], for_output_dataset=outdata)
# Add dataset to return dict
primary_datasets[name][designation] = primary_data
@@ -462,7 +468,9 @@ def collect_primary_datasets(job_context, output, input_ext):
if sa_session:
sa_session.add(outdata)
job_context.flush()
# Move discovered outputs to storage and set metdata / peeks
for callback in storage_callbacks:
callback()
return primary_datasets
lib/galaxy/jobs/actions/post.py
+10 -10
View File
@@ -209,7 +209,7 @@ class RenameDatasetAction(DefaultJobAction):
@classmethod
def execute(cls, app, sa_session, action, job, replacement_dict, final_job_state=None):
input_names = {}
# Lookp through inputs find one with "to_be_replaced" input
# Loop through inputs find one with "to_be_replaced" input
# variable name, and get the replacement name
for input_assoc in job.input_datasets:
if input_assoc.dataset:
@@ -399,31 +399,31 @@ class TagDatasetAction(DefaultJobAction):
@classmethod
def execute_on_mapped_over(cls, trans, sa_session, action, step_inputs, step_outputs, replacement_dict, final_job_state=None):
tag_handler = trans.app.tag_handler.create_tag_handler_session()
if action.action_arguments:
tags = [t.replace('#', 'name:') if t.startswith('#') else t for t in [t.strip() for t in action.action_arguments.get('tags', '').split(',') if t.strip()]]
if tags:
for name, step_output in step_outputs.items():
if action.output_name == '' or name == action.output_name:
cls._execute(trans.app, trans.user, step_output, tags)
cls._execute(tag_handler, trans.user, step_output, tags)
@classmethod
def execute(cls, app, sa_session, action, job, replacement_dict, final_job_state=None):
if action.action_arguments:
tag_handler = app.tag_handler.create_tag_handler_session()
tags = [t.replace('#', 'name:') if t.startswith('#') else t for t in [t.strip() for t in action.action_arguments.get('tags', '').split(',') if t.strip()]]
if tags:
for dataset_assoc in job.output_datasets:
if action.output_name == '' or dataset_assoc.name == action.output_name:
cls._execute(app, job.user, dataset_assoc.dataset, tags)
cls._execute(tag_handler, job.user, dataset_assoc.dataset, tags)
for dataset_collection_assoc in job.output_dataset_collection_instances:
if action.output_name == '' or dataset_collection_assoc.name == action.output_name:
cls._execute(app, job.user, dataset_collection_assoc.dataset_collection_instance, tags)
sa_session.flush()
cls._execute(tag_handler, job.user, dataset_collection_assoc.dataset_collection_instance, tags)
@classmethod
def _execute(cls, app, user, output, tags):
app.tag_handler.add_tags_from_list(user, output, tags)
def _execute(cls, tag_handler, user, output, tags):
tag_handler.add_tags_from_list(user, output, tags, flush=False)
@classmethod
def get_short_str(cls, pja):
@@ -443,8 +443,8 @@ class RemoveTagDatasetAction(TagDatasetAction):
direction = "from"
@classmethod
def _execute(cls, app, user, output, tags):
app.tag_handler.remove_tags_from_list(user, output, tags)
def _execute(cls, tag_handler, user, output, tags):
tag_handler.remove_tags_from_list(user, output, tags)
class ActionBox:
lib/galaxy/managers/collections.py
+8 -6
View File
@@ -57,7 +57,7 @@ class DatasetCollectionManager:
self.hda_manager = hda_manager
self.history_manager = history_manager
self.tag_handler = tag_handler
self.tag_handler = tag_handler.create_tag_handler_session()
self.ldda_manager = ldda_manager
def precreate_dataset_collection_instance(self, trans, parent, name, structure, implicit_inputs=None, implicit_output_name=None, tags=None, completed_collection=None):
@@ -182,7 +182,7 @@ class DatasetCollectionManager:
# values.
if isinstance(tags, list):
assert implicit_inputs is None, implicit_inputs
tags = self.tag_handler.add_tags_from_list(trans.user, dataset_collection_instance, tags)
tags = self.tag_handler.add_tags_from_list(trans.user, dataset_collection_instance, tags, flush=False)
else:
tags = self._append_tags(dataset_collection_instance, implicit_inputs, tags)
return self.__persist(dataset_collection_instance, flush=flush)
@@ -208,6 +208,8 @@ class DatasetCollectionManager:
hide_source_items=hide_source_items,
copy_elements=copy_elements,
history=history)
if history:
history.add_pending_items()
else:
if has_subcollections:
# Nested collection - recursively create collections as needed.
@@ -339,7 +341,7 @@ class DatasetCollectionManager:
copy_kwds["element_destination"] = parent # e.g. a history
if dataset_instance_attributes is not None:
copy_kwds["dataset_instance_attributes"] = dataset_instance_attributes
new_hdca = source_hdca.copy(**copy_kwds)
new_hdca = source_hdca.copy(flush=False, **copy_kwds)
new_hdca.copy_tags_from(target_user=trans.get_user(), source=source_hdca)
if not copy_elements:
parent.add_dataset_collection(new_hdca)
@@ -493,16 +495,16 @@ class DatasetCollectionManager:
decoded_id = int(trans.app.security.decode_id(encoded_id))
hda = self.hda_manager.get_accessible(decoded_id, trans.user)
if copy_elements:
element = self.hda_manager.copy(hda, history=history or trans.history, hide_copy=True)
element = self.hda_manager.copy(hda, history=history or trans.history, hide_copy=True, flush=False)
else:
element = hda
if hide_source_items and self.hda_manager.get_owned(hda.id, user=trans.user, current_history=history or trans.history):
hda.visible = False
self.tag_handler.apply_item_tags(user=trans.user, item=element, tags_str=tag_str)
self.tag_handler.apply_item_tags(user=trans.user, item=element, tags_str=tag_str, flush=False)
elif src_type == 'ldda':
element = self.ldda_manager.get(trans, encoded_id, check_accessible=True)
element = element.to_history_dataset_association(history or trans.history, add_to_history=True, visible=not hide_source_items)
self.tag_handler.apply_item_tags(user=trans.user, item=element, tags_str=tag_str)
self.tag_handler.apply_item_tags(user=trans.user, item=element, tags_str=tag_str, flush=False)
elif src_type == 'hdca':
# TODO: Option to copy? Force copy? Copy or allow if not owned?
element = self.__get_history_collection_instance(trans, encoded_id).collection
lib/galaxy/managers/hdas.py
+10 -10
View File
@@ -8,6 +8,8 @@ import gettext
import logging
import os
from sqlalchemy.orm.session import object_session
from galaxy import (
datatypes,
exceptions,
@@ -110,26 +112,24 @@ class HDAManager(datasets.DatasetAssociationManager,
self.session().flush()
return hda
def copy(self, hda, history=None, hide_copy=False, **kwargs):
def copy(self, hda, history=None, hide_copy=False, flush=True, **kwargs):
"""
Copy hda, including annotation and tags, add to history and return the given HDA.
"""
copy = hda.copy(parent_id=kwargs.get('parent_id'), copy_hid=False)
copy = hda.copy(parent_id=kwargs.get('parent_id'), copy_hid=False, copy_tags=hda.tags, flush=flush)
if hide_copy:
copy.visible = False
# add_dataset will update the hid to the next avail. in history
if history:
history.add_dataset(copy)
history.stage_addition(copy)
copy.copied_from_history_dataset_association = hda
copy.set_size()
original_annotation = self.annotation(hda)
self.annotate(copy, original_annotation, user=hda.history.user)
# these use a session flush
original_tags = self.get_tags(hda)
self.set_tags(copy, original_tags, user=hda.history.user)
self.annotate(copy, original_annotation, user=hda.history.user, flush=False)
if flush:
if history:
history.add_pending_items()
object_session(copy).flush()
return copy
lib/galaxy/model/__init__.py
+4 -6
View File
@@ -2420,10 +2420,9 @@ class History(Base, HasTags, Dictifiable, UsesAnnotations, HasName, Serializable
def stage_addition(self, items):
history_id = self.id
for item in listify(items):
item.history = self
if history_id:
item.history_id = history_id
else:
item.history = self
self._pending_additions.append(item)
@property
@@ -2568,10 +2567,9 @@ class History(Base, HasTags, Dictifiable, UsesAnnotations, HasName, Serializable
else:
hdcas = self.active_dataset_collections
for hdca in hdcas:
new_hdca = hdca.copy()
new_hdca = hdca.copy(flush=False)
new_history.add_dataset_collection(new_hdca, set_hid=False)
db_session.add(new_hdca)
db_session.flush()
if target_user:
new_hdca.copy_item_annotation(db_session, self.user, hdca, target_user, new_hdca)
@@ -5666,7 +5664,7 @@ class HistoryDatasetCollectionAssociation(
break
return matching_collection
def copy(self, element_destination=None, dataset_instance_attributes=None):
def copy(self, element_destination=None, dataset_instance_attributes=None, flush=True):
"""
Create a copy of this history dataset collection association. Copy
underlying collection.
@@ -5696,7 +5694,7 @@ class HistoryDatasetCollectionAssociation(
if element_destination:
element_destination.stage_addition(hdca)
element_destination.add_pending_items()
else:
if flush:
object_session(self).flush()
return hdca
lib/galaxy/model/base.py
+19
View File
@@ -2,6 +2,7 @@
Shared model and mapping code between Galaxy and Tool Shed, trying to
generalize to generic database connections.
"""
import os
import threading
from contextvars import ContextVar
from inspect import (
@@ -108,6 +109,24 @@ def versioned_objects(iter):
yield obj
def versioned_objects_strict(iter):
for obj in iter:
if hasattr(obj, '__create_version__'):
if obj.extension != "len":
# TODO: Custom builds (with .len extension) do not get a history or a HID.
# These should get some other type of permanent storage, perhaps UserDatasetAssociation ?
# Everything else needs to have a hid and a history
if not obj.history and not obj.history_id:
raise Exception(f'HistoryDatsetAssociation {obj} without history detected, this is not valid')
elif not obj.hid:
raise Exception(f'HistoryDatsetAssociation {obj} without has no hid, this is not valid')
yield obj
if os.environ.get("GALAXY_TEST_RAISE_EXCEPTION_ON_HISTORYLESS_HDA"):
versioned_objects = versioned_objects_strict # noqa: F811
def versioned_session(session):
@event.listens_for(session, 'before_flush')
def before_flush(session, flush_context, instances):
lib/galaxy/model/mapping.py
+7 -13
View File
@@ -45,19 +45,13 @@ def db_next_hid(self, n=1):
"""
session = object_session(self)
table = self.table
trans = session.begin()
try:
if "postgres" not in session.bind.dialect.name:
next_hid = select([table.c.hid_counter], table.c.id == model.cached_id(self)).with_for_update().scalar()
table.update(table.c.id == self.id).execute(hid_counter=(next_hid + n))
else:
stmt = table.update().where(table.c.id == model.cached_id(self)).values(hid_counter=(table.c.hid_counter + n)).returning(table.c.hid_counter)
next_hid = session.execute(stmt).scalar() - n
trans.commit()
return next_hid
except Exception:
trans.rollback()
raise
if "postgres" not in session.bind.dialect.name:
next_hid = select([table.c.hid_counter], table.c.id == model.cached_id(self)).with_for_update().scalar()
table.update(table.c.id == self.id).execute(hid_counter=(next_hid + n))
else:
stmt = table.update().where(table.c.id == model.cached_id(self)).values(hid_counter=(table.c.hid_counter + n)).returning(table.c.hid_counter)
next_hid = session.execute(stmt).scalar() - n
return next_hid
model.History._next_hid = db_next_hid # type: ignore
lib/galaxy/model/store/__init__.py
+1 -10
View File
@@ -275,7 +275,6 @@ class ModelImportStore(metaclass=abc.ABCMeta):
setattr(hda, attribute, value)
handle_dataset_object_edit(hda)
self._flush()
else:
metadata = dataset_attrs['metadata']
@@ -324,7 +323,6 @@ class ModelImportStore(metaclass=abc.ABCMeta):
dataset_instance.dataset.uuid = dataset_attrs["dataset_uuid"]
self._session_add(dataset_instance)
self._flush()
if model_class == "HistoryDatasetAssociation":
hda = cast(model.HistoryDatasetAssociation, dataset_instance)
@@ -336,7 +334,6 @@ class ModelImportStore(metaclass=abc.ABCMeta):
else:
object_import_tracker.requires_hid.append(hda)
self._flush()
file_source_root = self.file_source_root
# If dataset is in the dictionary - we will assert this dataset is tied to the Galaxy instance
@@ -447,7 +444,6 @@ class ModelImportStore(metaclass=abc.ABCMeta):
library_folder.deleted = deleted
self._session_add(library_folder)
self._flush()
for sub_folder_attrs in folder_attrs.get("folders", []):
sub_folder = import_folder(sub_folder_attrs)
@@ -465,8 +461,7 @@ class ModelImportStore(metaclass=abc.ABCMeta):
ld.library_dataset_dataset_association = ldda
self._session_add(ld)
self._flush()
self.sa_session.flush()
return library_folder
if 'root_folder' in library_attrs:
@@ -682,7 +677,6 @@ class ModelImportStore(metaclass=abc.ABCMeta):
assert not self.sessionless
job = self.sa_session.query(model.Job).get(job_attrs["id"])
self._connect_job_io(job, job_attrs, _find_hda, _find_hdca, _find_dce)
self._flush()
continue
imported_job = model.Job()
@@ -716,7 +710,6 @@ class ModelImportStore(metaclass=abc.ABCMeta):
except Exception:
pass
self._session_add(imported_job)
self._flush()
# Connect jobs to input and output datasets.
params = self._normalize_job_parameters(imported_job, job_attrs, _find_hda, _find_hdca, _find_dce)
@@ -725,7 +718,6 @@ class ModelImportStore(metaclass=abc.ABCMeta):
imported_job.add_parameter(name, dumps(value))
self._connect_job_io(imported_job, job_attrs, _find_hda, _find_hdca, _find_dce)
self._flush()
if object_key in job_attrs:
object_import_tracker.jobs_by_key[job_attrs[object_key]] = imported_job
@@ -747,7 +739,6 @@ class ModelImportStore(metaclass=abc.ABCMeta):
self._session_add(icja)
self._session_add(icj)
self._flush()
def _session_add(self, obj):
self.sa_session.add(obj)
lib/galaxy/model/store/discover.py
+27 -30
View File
@@ -59,6 +59,7 @@ class ModelPersistenceContext(metaclass=abc.ABCMeta):
created_from_basename=None,
final_job_state='ok',
creating_job_id=None,
storage_callbacks=None,
):
tag_list = tag_list or []
sources = sources or []
@@ -129,31 +130,8 @@ class ModelPersistenceContext(metaclass=abc.ABCMeta):
if created_from_basename is not None:
primary_data.created_from_basename = created_from_basename
has_flushed = False
if tag_list:
# If we have a tag we need a primary id, so need to flush here
# TODO: eliminate creating tag associations within create dataset
# We can do this incrementally by not passing in a tag list.
self.flush()
has_flushed = True
self.tag_handler.add_tags_from_list(self.job.user, primary_data, tag_list)
# Move data from temp location to dataset location
if filename:
# TODO: eliminate this, should happen outside of create_dataset so that we don't need to flush
if not has_flushed:
self.flush()
has_flushed = True
if not link_data:
self.object_store.update_from_file(primary_data.dataset, file_name=filename, create=True)
else:
primary_data.link_to(filename)
if extra_files:
persist_extra_files(self.object_store, extra_files, primary_data)
primary_data.set_size()
else:
# We are sure there are no extra files, so optimize things that follow by settting total size also.
primary_data.set_size(no_extra_files=True)
self.tag_handler.add_tags_from_list(self.job.user, primary_data, tag_list, flush=False)
# If match specified a name use otherwise generate one from
# designation.
@@ -174,12 +152,28 @@ class ModelPersistenceContext(metaclass=abc.ABCMeta):
if info is not None:
primary_data.info = info
if filename:
self.set_datasets_metadata(datasets=[primary_data], datasets_attributes=[dataset_attributes])
if storage_callbacks is None:
self.finalize_storage(primary_data=primary_data, dataset_attributes=dataset_attributes, extra_files=extra_files, filename=filename, link_data=link_data)
else:
storage_callbacks.append(lambda: self.finalize_storage(primary_data=primary_data, dataset_attributes=dataset_attributes, extra_files=extra_files, filename=filename, link_data=link_data))
return primary_data
def finalize_storage(self, primary_data, dataset_attributes, extra_files, filename, link_data):
# Move data from temp location to dataset location
if not link_data:
self.object_store.update_from_file(primary_data.dataset, file_name=filename, create=True)
else:
primary_data.link_to(filename)
if extra_files:
persist_extra_files(self.object_store, extra_files, primary_data)
primary_data.set_size()
else:
# We are sure there are no extra files, so optimize things that follow by settting total size also.
primary_data.set_size(no_extra_files=True)
# TODO: this might run set_meta after copying the file to the object store, which could be inefficient if job working directory is closer to the node.
self.set_datasets_metadata(datasets=[primary_data], datasets_attributes=[dataset_attributes])
@staticmethod
def set_datasets_metadata(datasets, datasets_attributes=None):
datasets_attributes = datasets_attributes or [{} for _ in datasets]
@@ -297,9 +291,9 @@ class ModelPersistenceContext(metaclass=abc.ABCMeta):
association_name = f'__new_primary_file_{name}|{element_identifier_str}__'
self.add_output_dataset_association(association_name, dataset)
self.update_object_store_with_datasets(datasets=element_datasets['datasets'], paths=element_datasets['paths'], extra_files=element_datasets['extra_files'])
add_datasets_timer = ExecutionTimer()
self.add_datasets_to_history(element_datasets['datasets'])
self.update_object_store_with_datasets(datasets=element_datasets['datasets'], paths=element_datasets['paths'], extra_files=element_datasets['extra_files'])
log.debug(
"(%s) Add dynamic collection datasets to history for output [%s] %s",
self.job_id(),
@@ -314,9 +308,8 @@ class ModelPersistenceContext(metaclass=abc.ABCMeta):
# that's not better or worse than what we previously did in create_datasets
# TDOD: implement that or figure out why it is not implemented and find a better solution.
# Could it be that SessionlessModelPersistenceContext doesn't support tags?
tag_session = self.tag_handler.create_tag_handler_session()
for dataset, tags in zip(datasets, tag_lists):
tag_session.add_tags_from_list(self.job.user, dataset, tags, flush=False)
self.tag_handler.add_tags_from_list(self.job.user, dataset, tags, flush=False)
def update_object_store_with_datasets(self, datasets, paths, extra_files):
for dataset, path, extra_file in zip(datasets, paths, extra_files):
@@ -603,6 +596,7 @@ def persist_elements_to_folder(model_persistence_context, elements, library_fold
def persist_hdas(elements, model_persistence_context, final_job_state='ok'):
# discover files as individual datasets for the target history
datasets = []
storage_callbacks = []
def collect_elements_for_history(elements):
for element in elements:
@@ -648,12 +642,15 @@ def persist_hdas(elements, model_persistence_context, final_job_state='ok'):
hashes=hashes,
created_from_basename=created_from_basename,
final_job_state=state,
storage_callbacks=storage_callbacks,
)
if not hda_id:
datasets.append(dataset)
collect_elements_for_history(elements)
model_persistence_context.add_datasets_to_history(datasets)
for callback in storage_callbacks:
callback()
def add_datasets_to_history(self, datasets, for_output_dataset=None):
if for_output_dataset is not None:
lib/galaxy/tool_util/linters/inputs.py
+11 -13
View File
@@ -19,16 +19,16 @@ FILTER_TYPES = [
ATTRIB_VALIDATOR_COMPATIBILITY = {
"check": ["metadata"],
"expression": ["regex", "substitute_value_in_message"],
"expression": ["substitute_value_in_message"],
"table_name": ["dataset_metadata_in_data_table", "dataset_metadata_not_in_data_table", "value_in_data_table", "value_not_in_data_table"],
"filename": ["dataset_metadata_in_file"],
"metadata_name": ["dataset_metadata_in_data_table", "dataset_metadata_not_in_data_table", "dataset_metadata_in_file"],
"metadata_column": ["dataset_metadata_in_data_table", "dataset_metadata_not_in_data_table", "value_in_data_table", "value_not_in_data_table", "dataset_metadata_in_file options"],
"metadata_column": ["dataset_metadata_in_data_table", "dataset_metadata_not_in_data_table", "value_in_data_table", "value_not_in_data_table", "dataset_metadata_in_file"],
"line_startswith": ["dataset_metadata_in_file"],
"min": ["in_range", "length"],
"max": ["in_range", "length"],
"exclude_min": ["in_range"],
"exclude_max": ["in_range"],
"min": ["in_range", "length", "dataset_metadata_in_range"],
"max": ["in_range", "length", "dataset_metadata_in_range"],
"exclude_min": ["in_range", "dataset_metadata_in_range"],
"exclude_max": ["in_range", "dataset_metadata_in_range"],
"split": ["dataset_metadata_in_file"],
"skip": ["metadata"]
}
@@ -132,7 +132,7 @@ def lint_inputs(tool_xml, lint_ctx):
# lint statically defined options
if any(['value' not in option.attrib for option in select_options]):
lint_ctx.error(f"Select parameter [{param_name}] has option without value")
if len(set([option.text.strip() for option in select_options])) != len(select_options):
if len(set([option.text.strip() for option in select_options if option.text is not None])) != len(select_options):
lint_ctx.error(f"Select parameter [{param_name}] has multiple options with the same text content")
if len(set([option.attrib.get("value") for option in select_options])) != len(select_options):
lint_ctx.error(f"Select parameter [{param_name}] has multiple options with the same value")
@@ -161,14 +161,12 @@ def lint_inputs(tool_xml, lint_ctx):
lint_ctx.error(f"Parameter [{param_name}]: attribute '{attrib}' is incompatible with validator of type '{vtype}'")
if vtype == "expression" and validator.text is None:
lint_ctx.error(f"Parameter [{param_name}]: expression validator without content")
if vtype != "expression" and validator.text is not None:
if vtype not in ["expression", "regex"] and validator.text is not None:
lint_ctx.warn(f"Parameter [{param_name}]: '{vtype}' validators are not expected to contain text (found '{validator.text}')")
if vtype == "regex" and "expression" not in validator.attrib:
lint_ctx.error(f"Parameter [{param_name}]: '{vtype}' validators need to define an 'expression' attribute")
if vtype in ["in_range", "length", "dataset_metadata_in_range"] and ("min" not in validator.attrib or "max" not in validator.attrib):
if vtype in ["in_range", "length", "dataset_metadata_in_range"] and ("min" not in validator.attrib and "max" not in validator.attrib):
lint_ctx.error(f"Parameter [{param_name}]: '{vtype}' validators need to define the 'min' or 'max' attribute(s)")
if vtype in ["metadata"] and ("check" not in validator.attrib or "skip" not in validator.attrib):
lint_ctx.error(f"Parameter [{param_name}]: '{vtype}' validators need to define the 'check' or 'skip' attribute(s)")
if vtype in ["metadata"] and ("check" not in validator.attrib and "skip" not in validator.attrib):
lint_ctx.error(f"Parameter [{param_name}]: '{vtype}' validators need to define the 'check' or 'skip' attribute(s) {validator.attrib}")
if vtype in ["value_in_data_table", "value_not_in_data_table", "dataset_metadata_in_data_table", "dataset_metadata_not_in_data_table"] and "table_name" not in validator.attrib:
lint_ctx.error(f"Parameter [{param_name}]: '{vtype}' validators need to define the 'table_name' attribute")
lib/galaxy/tools/actions/__init__.py
+10 -8
View File
@@ -84,8 +84,6 @@ class DefaultToolAction:
if formats is None:
formats = input.formats
# Need to refresh in case this conversion just took place, i.e. input above in tool performed the same conversion
trans.sa_session.refresh(data)
direct_match, target_ext, converted_dataset = data.find_conversion_destination(formats)
if not direct_match and target_ext:
if converted_dataset:
@@ -539,6 +537,7 @@ class DefaultToolAction:
for name, data in out_data.items():
if name not in child_dataset_names and name not in incoming: # don't add children; or already existing datasets, i.e. async created
history.stage_addition(data)
history.add_pending_items(set_output_hid=set_output_hid)
# Add all the children to their parents
for parent_name, child_name in parent_to_child_pairs:
@@ -560,13 +559,16 @@ class DefaultToolAction:
if completed_job:
job.set_copied_from_job_id(completed_job.id)
trans.sa_session.add(job)
# Now that we have a job id, we can remap any outputs if this is a rerun and the user chose to continue dependent jobs
# Remap any outputs if this is a rerun and the user chose to continue dependent jobs
# This functionality requires tracking jobs in the database.
if app.config.track_jobs_in_database and rerun_remap_job_id is not None:
# We need a flush here and get hids in order to rewrite jobs parameter,
# but remapping jobs should only affect single jobs anyway, so this is not too costly.
history.add_pending_items(set_output_hid=set_output_hid)
trans.sa_session.flush()
# Need to flush here so that referencing outputs by id works
session = trans.sa_session()
try:
session.expire_on_commit = False
session.flush()
finally:
session.expire_on_commit = True
self._remap_job_on_rerun(trans=trans,
galaxy_session=galaxy_session,
rerun_remap_job_id=rerun_remap_job_id,
@@ -597,7 +599,6 @@ class DefaultToolAction:
else:
if flush_job:
# Set HID and add to history.
history.add_pending_items(set_output_hid=set_output_hid)
job_flush_timer = ExecutionTimer()
trans.sa_session.flush()
log.info(f"Flushed transaction for job {job.log_str()} {job_flush_timer}")
@@ -820,6 +821,7 @@ class OutputCollections:
def __init__(self, trans, history, tool, tool_action, input_collections, dataset_collection_elements, on_text, incoming, params, job_params, tags, hdca_tags):
self.trans = trans
self.tag_handler = trans.app.tag_handler.create_tag_handler_session()
self.history = history
self.tool = tool
self.tool_action = tool_action
lib/galaxy/tools/actions/upload_common.py
+4 -4
View File
@@ -232,12 +232,12 @@ def __new_library_upload(trans, cntrller, uploaded_dataset, library_bunch, tag_h
sa_session=trans.sa_session)
if uploaded_dataset.get('tag_using_filenames', False):
tag_from_filename = os.path.splitext(os.path.basename(uploaded_dataset.name))[0]
tag_handler.apply_item_tag(item=ldda, user=trans.user, name='name', value=tag_from_filename)
tag_handler.apply_item_tag(item=ldda, user=trans.user, name='name', value=tag_from_filename, flush=False)
tags_list = uploaded_dataset.get('tags', False)
if tags_list:
for tag in tags_list:
tag_handler.apply_item_tag(item=ldda, user=trans.user, name='name', value=tag)
tag_handler.apply_item_tag(item=ldda, user=trans.user, name='name', value=tag, flush=False)
trans.sa_session.add(ldda)
if state:
@@ -294,12 +294,12 @@ def new_upload(trans, cntrller, uploaded_dataset, library_bunch=None, history=No
if library_bunch.tags and not uploaded_dataset.tags:
new_tags = tag_handler.parse_tags_list(library_bunch.tags)
for tag in new_tags:
tag_handler.apply_item_tag(user=trans.user, item=upload_target_dataset_instance, name=tag[0], value=tag[1])
tag_handler.apply_item_tag(user=trans.user, item=upload_target_dataset_instance, name=tag[0], value=tag[1], flush=False)
else:
upload_target_dataset_instance = __new_history_upload(trans, uploaded_dataset, history=history, state=state)
if tag_list:
tag_handler.add_tags_from_list(trans.user, upload_target_dataset_instance, tag_list)
tag_handler.add_tags_from_list(trans.user, upload_target_dataset_instance, tag_list, flush=False)
return upload_target_dataset_instance
lib/galaxy/tools/execute.py
+5 -7
View File
@@ -111,18 +111,16 @@ def execute(trans, tool, mapping_params, history, rerun_remap_job_id=None, colle
history = execution_slice.history or history
jobs_executed += 1
if execution_slice:
# a side effect of adding datasets to a history is a commit within db_next_hid (even with flush=False).
history.add_pending_items()
else:
# Make sure collections, implicit jobs etc are flushed even if there are no precreated output datasets
trans.sa_session.flush()
if job_datasets:
for job, datasets in job_datasets.items():
for dataset_instance in datasets:
dataset_instance.dataset.job = job
if execution_slice:
history.add_pending_items()
# Make sure collections, implicit jobs etc are flushed even if there are no precreated output datasets
trans.sa_session.flush()
tool_id = tool.id
for job in execution_tracker.successful_jobs:
# Put the job in the queue if tracking in memory
lib/galaxy/webapps/galaxy/api/dataset_collections.py
+2 -1
View File
@@ -42,7 +42,7 @@ class DatasetCollectionsController(
:type payload: dict
:param payload: (optional) dictionary structure containing:
* collection_type: dataset colltion type to create.
* collection_type: dataset collection type to create.
* instance_type: Instance type - 'history' or 'library'.
* name: the new dataset collections's name
* datasets: object describing datasets for collection
@@ -57,6 +57,7 @@ class DatasetCollectionsController(
history_id = decode_id(self.app, history_id)
history = self.history_manager.get_owned(history_id, trans.user, current_history=trans.history)
create_params["parent"] = history
create_params["history"] = history
elif instance_type == "library":
folder_id = payload.get('folder_id')
library_folder = self.get_library_folder(trans, folder_id, check_accessible=True)
lib/galaxy/workflow/run_request.py
+3 -6
View File
@@ -326,11 +326,8 @@ def build_workflow_run_configs(trans, workflow, payload):
else:
raise exceptions.RequestParameterInvalidException(f"Unknown workflow input source '{input_source}' specified.")
if add_to_history and content.history != history:
content = content.copy()
if isinstance(content, app.model.HistoryDatasetAssociation):
history.add_dataset(content)
else:
history.add_dataset_collection(content)
content = content.copy(flush=False)
history.stage_addition(content)
input_dict['content'] = content
except AssertionError:
raise exceptions.ItemAccessibilityException(f"Invalid workflow input '{input_id}' specified")
@@ -368,7 +365,7 @@ def build_workflow_run_configs(trans, workflow, payload):
valid_option = True
if not valid_option:
raise exceptions.RequestParameterInvalidException(f"Invalid value for parameter '{name}' found.")
history.add_pending_items()
run_configs.append(WorkflowRunConfig(
target_history=history,
replacement_dict=payload.get('replacement_params', {}),
test/unit/app/jobs/test_job_context.py
+1
View File
@@ -55,6 +55,7 @@ def test_job_context_discover_outputs_flushes_once(mocker):
job = model.Job()
job.history = h
sa_session.add(job)
sa_session.flush()
job_working_directory = tempfile.mkdtemp()
setup_data(job_working_directory)
permission_provider = PermissionProvider()
test/unit/app/tools/test_history_imp_exp.py
+1
View File
@@ -535,6 +535,7 @@ def test_export_copied_objects_copied_outside_history():
other_h = model.History(name=h.name + "-other", user=h.user)
sa_session.add(other_h)
sa_session.flush()
hc3 = hc2.copy(element_destination=other_h)
other_h.add_pending_items()
test/unit/data/model/test_model_store.py
+2 -2
View File
@@ -228,10 +228,9 @@ def test_import_export_edit_collection():
sa_session.add(hc1)
sa_session.add(h)
sa_session.flush()
import_history = model.History(name="Test History for Import", user=u)
sa_session.add(import_history)
sa_session.flush()
temp_directory = mkdtemp()
with store.DirectoryModelExportStore(temp_directory, app=app, for_edit=True) as export_store:
@@ -411,6 +410,7 @@ def _setup_simple_export(export_kwds):
import_history = model.History(name="Test History for Import", user=u)
sa_session.add(import_history)
sa_session.flush()
temp_directory = mkdtemp()
with store.DirectoryModelExportStore(temp_directory, app=app, **export_kwds) as export_store:
test/unit/tool_util/test_tool_linters.py
+51 -2
View File
@@ -146,6 +146,51 @@ VALIDATOR_INCOMPATIBILITIES = """
</tool>
"""
VALIDATOR_CORRECT = """
<tool name="BWA Mapper" id="bwa" version="1.0.1" is_multi_byte="true" display_interface="true" require_login="true" hidden="true">
<description>The BWA Mapper</description>
<version_command interpreter="python">bwa.py --version</version_command>
<inputs>
<param name="data_param" type="data" format="data">
<validator type="metadata" check="md1,md2" skip="md3,md4" message="cutom validation message" negate="true"/>
<validator type="unspecified_build" message="cutom validation message" negate="true"/>
<validator type="dataset_ok_validator" message="cutom validation message" negate="true"/>
<validator type="dataset_metadata_in_range" min="0" max="100" exclude_min="true" exclude_max="true" message="cutom validation message" negate="true"/>
<validator type="dataset_metadata_in_file" filename="file.tsv" metadata_column="3" split="," message="cutom validation message" negate="true"/>
<validator type="dataset_metadata_in_data_table" table_name="datatable_name" metadata_column="3" message="cutom validation message" negate="true"/>
</param>
<param name="collection_param" type="collection">
<validator type="metadata" check="md1,md2" skip="md3,md4" message="cutom validation message"/>
<validator type="unspecified_build" message="cutom validation message"/>
<validator type="dataset_ok_validator" message="cutom validation message"/>
<validator type="dataset_metadata_in_range" min="0" max="100" exclude_min="true" exclude_max="true" message="cutom validation message"/>
<validator type="dataset_metadata_in_file" filename="file.tsv" metadata_column="3" split="," message="cutom validation message"/>
<validator type="dataset_metadata_in_data_table" table_name="datatable_name" metadata_column="3" message="cutom validation message"/>
</param>
<param name="text_param" type="text">
<validator type="regex">reg.xp</validator>
<validator type="length" min="0" max="100" message="cutom validation message"/>
<validator type="empty_field" message="cutom validation message"/>
<validator type="value_in_data_table" table_name="datatable_name" metadata_column="3" message="cutom validation message"/>
<validator type="expression" message="cutom validation message">somepythonexpression</validator>
</param>
<param name="select_param" type="select">
<options from_data_table="bowtie2_indexes"/>
<validator type="no_options" negate="true"/>
<validator type="regex" negate="true">reg.xp</validator>
<validator type="length" min="0" max="100" message="cutom validation message" negate="true"/>
<validator type="empty_field" message="cutom validation message" negate="true"/>
<validator type="value_in_data_table" table_name="datatable_name" metadata_column="3" message="cutom validation message" negate="true"/>
<validator type="expression" message="cutom validation message" negate="true">somepythonexpression</validator>
</param>
<param name="int_param" type="integer">
<validator type="in_range" min="0" max="100" exclude_min="true" exclude_max="true" negate="true"/>
<validator type="expression" message="cutom validation message">somepythonexpression</validator>
</param>
</inputs>
</tool>
"""
# check that linter accepts format source for collection elements as means to specify format
# and that the linter warns if format and format_source are used
OUTPUTS_COLLECTION_FORMAT_SOURCE = """
@@ -246,8 +291,11 @@ TESTS = [
and "Parameter [param_name]: validator with an incompatible type 'in_range'" in x.error_messages
and "Parameter [param_name]: 'in_range' validators need to define the 'min' or 'max' attribute(s)" in x.error_messages
and "Parameter [param_name]: attribute 'filename' is incompatible with validator of type 'regex'" in x.error_messages
and "Parameter [param_name]: 'regex' validators need to define an 'expression' attribute" in x.error_messages
and len(x.warn_messages) == 1 and len(x.error_messages) == 4
and len(x.warn_messages) == 1 and len(x.error_messages) == 3
),
(
VALIDATOR_CORRECT, inputs.lint_inputs,
lambda x: len(x.warn_messages) == 0 and len(x.error_messages) == 0
),
(
OUTPUTS_COLLECTION_FORMAT_SOURCE, outputs.lint_output,
@@ -272,6 +320,7 @@ TEST_IDS = [
'select deprecations',
'select option definitions',
'validator imcompatibilities',
'validator all correct',
'outputs collection static elements with format_source',
'outputs discover datatsets with tool provided metadata'
]