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 pull request #12756 from ic4f/dev_install_model_declarative

Browse Source 
Convert install_model to declarative mapping
This commit is contained in:
Marius van den Beek
2021-10-21 10:26:26 +02:00
committed by GitHub
parent a5bb3e9266 404bac3903
commit f302a2def7
26 changed files with 605 additions and 202 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/model/tool_shed_install/__init__.py
+126 -41
View File
@@ -3,6 +3,27 @@ import os
from enum import Enum
from typing import TYPE_CHECKING
from sqlalchemy import (
Boolean,
Column,
DateTime,
ForeignKey,
Integer,
String,
Table,
TEXT,
)
from sqlalchemy.orm import (
registry,
relationship,
)
from sqlalchemy.orm.decl_api import DeclarativeMeta
from galaxy.model.custom_types import (
MutableJSONType,
TrimmedString
)
from galaxy.model.orm.now import now
from galaxy.util import asbool
from galaxy.util.bunch import Bunch
from galaxy.util.dictifiable import Dictifiable
@@ -10,17 +31,53 @@ from galaxy.util.tool_shed import common_util
log = logging.getLogger(__name__)
mapper_registry = registry()
if TYPE_CHECKING:
from sqlalchemy.schema import Table
class _HasTable:
table: Table
else:
_HasTable = object
class ToolShedRepository(_HasTable):
class Base(metaclass=DeclarativeMeta):
__abstract__ = True
registry = mapper_registry
metadata = mapper_registry.metadata
__init__ = mapper_registry.constructor
@classmethod
def __declare_last__(cls):
cls.table = cls.__table__
class ToolShedRepository(Base, _HasTable):
__tablename__ = 'tool_shed_repository'
id = Column(Integer, primary_key=True)
create_time = Column(DateTime, default=now)
update_time = Column(DateTime, default=now, onupdate=now)
tool_shed = Column(TrimmedString(255), index=True)
name = Column(TrimmedString(255), index=True)
description = Column(TEXT)
owner = Column(TrimmedString(255), index=True)
installed_changeset_revision = Column(TrimmedString(255))
changeset_revision = Column(TrimmedString(255), index=True)
ctx_rev = Column(TrimmedString(10))
metadata_ = Column('metadata', MutableJSONType, nullable=True)
includes_datatypes = Column(Boolean, index=True, default=False)
tool_shed_status = Column(MutableJSONType, nullable=True)
deleted = Column(Boolean, index=True, default=False)
uninstalled = Column(Boolean, default=False)
dist_to_shed = Column(Boolean, default=False)
status = Column(TrimmedString(255))
error_message = Column(TEXT)
tool_versions = relationship('ToolVersion', back_populates='tool_shed_repository')
tool_dependencies = relationship('ToolDependency', order_by='ToolDependency.name',
back_populates='tool_shed_repository')
required_repositories = relationship('RepositoryRepositoryDependencyAssociation',
back_populates='repository')
dict_collection_visible_keys = ['id', 'tool_shed', 'name', 'owner', 'installed_changeset_revision', 'changeset_revision', 'ctx_rev', 'includes_datatypes',
'tool_shed_status', 'deleted', 'uninstalled', 'dist_to_shed', 'status', 'error_message', 'description']
dict_element_visible_keys = ['id', 'tool_shed', 'name', 'owner', 'installed_changeset_revision', 'changeset_revision', 'ctx_rev', 'includes_datatypes',
@@ -46,7 +103,7 @@ class ToolShedRepository(_HasTable):
UNINSTALLED = 'deleted_new'
def __init__(self, id=None, create_time=None, tool_shed=None, name=None, description=None, owner=None, installed_changeset_revision=None,
changeset_revision=None, ctx_rev=None, metadata=None, includes_datatypes=False, tool_shed_status=None, deleted=False,
changeset_revision=None, ctx_rev=None, metadata_=None, includes_datatypes=False, tool_shed_status=None, deleted=False,
uninstalled=False, dist_to_shed=False, status=None, error_message=None):
self.id = id
self.create_time = create_time
@@ -57,7 +114,7 @@ class ToolShedRepository(_HasTable):
self.installed_changeset_revision = installed_changeset_revision
self.changeset_revision = changeset_revision
self.ctx_rev = ctx_rev
self.metadata = metadata or {}
self.metadata_ = metadata_ or {}
self.includes_datatypes = includes_datatypes
self.tool_shed_status = tool_shed_status
self.deleted = deleted
@@ -96,11 +153,11 @@ class ToolShedRepository(_HasTable):
@property
def shed_config_filename(self):
return self.metadata.get('shed_config_filename')
return self.metadata_.get('shed_config_filename')
@shed_config_filename.setter
def shed_config_filename(self, value):
self.metadata['shed_config_filename'] = os.path.abspath(value)
self.metadata_['shed_config_filename'] = os.path.abspath(value)
def get_shed_config_dict(self, app):
"""
@@ -125,7 +182,7 @@ class ToolShedRepository(_HasTable):
def guess_shed_config(self, app):
tool_ids = []
for tool in self.metadata.get('tools', []):
for tool in self.metadata_.get('tools', []):
tool_ids.append(tool.get('guid'))
for shed_tool_conf_dict in app.toolbox.dynamic_confs(include_migrated_tool_conf=True):
name = shed_tool_conf_dict['config_filename']
@@ -158,11 +215,11 @@ class ToolShedRepository(_HasTable):
@property
def has_readme_files(self):
return 'readme_files' in self.metadata
return 'readme_files' in self.metadata_
@property
def has_repository_dependencies(self):
repository_dependencies_dict = self.metadata.get('repository_dependencies', {})
repository_dependencies_dict = self.metadata_.get('repository_dependencies', {})
repository_dependencies = repository_dependencies_dict.get('repository_dependencies', [])
# [["http://localhost:9009", "package_libgtextutils_0_6", "test", "e2003cbf18cd", "True", "True"]]
for rd_tup in repository_dependencies:
@@ -174,7 +231,7 @@ class ToolShedRepository(_HasTable):
@property
def has_repository_dependencies_only_if_compiling_contained_td(self):
repository_dependencies_dict = self.metadata.get('repository_dependencies', {})
repository_dependencies_dict = self.metadata_.get('repository_dependencies', {})
repository_dependencies = repository_dependencies_dict.get('repository_dependencies', [])
# [["http://localhost:9009", "package_libgtextutils_0_6", "test", "e2003cbf18cd", "True", "True"]]
for rd_tup in repository_dependencies:
@@ -190,16 +247,16 @@ class ToolShedRepository(_HasTable):
@property
def includes_data_managers(self):
return bool(len(self.metadata.get('data_manager', {}).get('data_managers', {})))
return bool(len(self.metadata_.get('data_manager', {}).get('data_managers', {})))
@property
def includes_tools(self):
return 'tools' in self.metadata
return 'tools' in self.metadata_
@property
def includes_tools_for_display_in_tool_panel(self):
if self.includes_tools:
tool_dicts = self.metadata['tools']
tool_dicts = self.metadata_['tools']
for tool_dict in tool_dicts:
if tool_dict.get('add_to_tool_panel', True):
return True
@@ -207,11 +264,11 @@ class ToolShedRepository(_HasTable):
@property
def includes_tool_dependencies(self):
return 'tool_dependencies' in self.metadata
return 'tool_dependencies' in self.metadata_
@property
def includes_workflows(self):
return 'workflows' in self.metadata
return 'workflows' in self.metadata_
@property
def installed_repository_dependencies(self):
@@ -350,7 +407,7 @@ class ToolShedRepository(_HasTable):
"""
required_rd_tups_that_must_be_installed = []
if self.has_repository_dependencies:
rd_tups = self.metadata['repository_dependencies']['repository_dependencies']
rd_tups = self.metadata_['repository_dependencies']['repository_dependencies']
for rd_tup in rd_tups:
if len(rd_tup) == 5:
tool_shed, name, owner, changeset_revision, prior_installation_required, only_if_compiling_contained_td = \
@@ -447,7 +504,7 @@ class ToolShedRepository(_HasTable):
dependencies.
"""
rd_tups_of_repositories_needed_for_compiling_td = []
repository_dependencies = self.metadata.get('repository_dependencies', {})
repository_dependencies = self.metadata_.get('repository_dependencies', {})
rd_tups = repository_dependencies.get('repository_dependencies', [])
for rd_tup in rd_tups:
if len(rd_tup) == 6:
@@ -484,20 +541,49 @@ class ToolShedRepository(_HasTable):
return False
class RepositoryRepositoryDependencyAssociation(_HasTable):
class RepositoryRepositoryDependencyAssociation(Base, _HasTable):
__tablename__ = 'repository_repository_dependency_association'
id = Column(Integer, primary_key=True)
create_time = Column(DateTime, default=now)
update_time = Column(DateTime, default=now, onupdate=now)
tool_shed_repository_id = Column(ForeignKey('tool_shed_repository.id'), index=True)
repository_dependency_id = Column(ForeignKey('repository_dependency.id'), index=True)
repository = relationship('ToolShedRepository', back_populates='required_repositories')
repository_dependency = relationship('RepositoryDependency')
def __init__(self, tool_shed_repository_id=None, repository_dependency_id=None):
self.tool_shed_repository_id = tool_shed_repository_id
self.repository_dependency_id = repository_dependency_id
class RepositoryDependency(_HasTable):
class RepositoryDependency(Base, _HasTable):
__tablename__ = 'repository_dependency'
id = Column(Integer, primary_key=True)
create_time = Column(DateTime, default=now)
update_time = Column(DateTime, default=now, onupdate=now)
tool_shed_repository_id = Column(ForeignKey('tool_shed_repository.id'), index=True, nullable=False)
repository = relationship('ToolShedRepository')
def __init__(self, tool_shed_repository_id=None):
self.tool_shed_repository_id = tool_shed_repository_id
class ToolDependency(_HasTable):
class ToolDependency(Base, _HasTable):
__tablename__ = 'tool_dependency'
id = Column(Integer, primary_key=True)
create_time = Column(DateTime, default=now)
update_time = Column(DateTime, default=now, onupdate=now)
tool_shed_repository_id = Column(ForeignKey('tool_shed_repository.id'), index=True, nullable=False)
name = Column(TrimmedString(255))
version = Column(TEXT)
type = Column(TrimmedString(40))
status = Column(TrimmedString(255), nullable=False)
error_message = Column(TEXT)
tool_shed_repository = relationship('ToolShedRepository', back_populates='tool_dependencies')
# converting this one to Enum breaks the tool shed tests,
# don't know why though -John
installation_status = Bunch(NEVER_INSTALLED='Never installed',
@@ -568,14 +654,21 @@ class ToolDependency(_HasTable):
return self.status == self.installation_status.INSTALLED
class ToolVersion(Dictifiable, _HasTable):
dict_element_visible_keys = ['id', 'tool_shed_repository']
class ToolVersion(Base, Dictifiable, _HasTable):
__tablename__ = 'tool_version'
def __init__(self, id=None, create_time=None, tool_id=None, tool_shed_repository=None):
self.id = id
self.create_time = create_time
self.tool_id = tool_id
self.tool_shed_repository = tool_shed_repository
id = Column(Integer, primary_key=True)
create_time = Column(DateTime, default=now)
update_time = Column(DateTime, default=now, onupdate=now)
tool_id = Column(String(255))
tool_shed_repository_id = Column(ForeignKey('tool_shed_repository.id'), index=True, nullable=True)
parent_tool_association = relationship('ToolVersionAssociation',
primaryjoin=(lambda: ToolVersion.id == ToolVersionAssociation.tool_id)) # type: ignore
child_tool_association = relationship('ToolVersionAssociation',
primaryjoin=(lambda: ToolVersion.id == ToolVersionAssociation.parent_id)) # type: ignore
tool_shed_repository = relationship('ToolShedRepository', back_populates='tool_versions')
dict_element_visible_keys = ['id', 'tool_shed_repository']
def to_dict(self, view='element'):
rval = super().to_dict(view=view)
@@ -587,17 +680,9 @@ class ToolVersion(Dictifiable, _HasTable):
return rval
class ToolVersionAssociation(_HasTable):
class ToolVersionAssociation(Base, _HasTable):
__tablename__ = 'tool_version_association'
def __init__(self, id=None, tool_id=None, parent_id=None):
self.id = id
self.tool_id = tool_id
self.parent_id = parent_id
class MigrateTools(_HasTable):
def __init__(self, repository_id=None, repository_path=None, version=None):
self.repository_id = repository_id
self.repository_path = repository_path
self.version = version
id = Column(Integer, primary_key=True)
tool_id = Column(ForeignKey('tool_version.id'), index=True, nullable=False)
parent_id = Column(ForeignKey('tool_version.id'), index=True, nullable=False)
lib/galaxy/model/tool_shed_install/mapping.py
+2 -115
View File
@@ -1,122 +1,9 @@
from sqlalchemy import (
Boolean,
Column,
DateTime,
ForeignKey,
Integer,
MetaData,
String,
Table,
TEXT
)
from sqlalchemy.orm import (
mapper,
relation
)
from galaxy.model import tool_shed_install as install_model
from galaxy.model.base import ModelMapping
from galaxy.model.custom_types import (
MutableJSONType,
TrimmedString
)
from galaxy.model.orm.engine_factory import build_engine
from galaxy.model.orm.now import now
from galaxy.model.tool_shed_install import mapper_registry
metadata = MetaData()
install_model.ToolShedRepository.table = Table("tool_shed_repository", metadata,
Column("id", Integer, primary_key=True),
Column("create_time", DateTime, default=now),
Column("update_time", DateTime, default=now, onupdate=now),
Column("tool_shed", TrimmedString(255), index=True),
Column("name", TrimmedString(255), index=True),
Column("description", TEXT),
Column("owner", TrimmedString(255), index=True),
Column("installed_changeset_revision", TrimmedString(255)),
Column("changeset_revision", TrimmedString(255), index=True),
Column("ctx_rev", TrimmedString(10)),
Column("metadata", MutableJSONType, nullable=True),
Column("includes_datatypes", Boolean, index=True, default=False),
Column("tool_shed_status", MutableJSONType, nullable=True),
Column("deleted", Boolean, index=True, default=False),
Column("uninstalled", Boolean, default=False),
Column("dist_to_shed", Boolean, default=False),
Column("status", TrimmedString(255)),
Column("error_message", TEXT))
install_model.RepositoryRepositoryDependencyAssociation.table = Table('repository_repository_dependency_association', metadata,
Column("id", Integer, primary_key=True),
Column("create_time", DateTime, default=now),
Column("update_time", DateTime, default=now, onupdate=now),
Column("tool_shed_repository_id", Integer, ForeignKey("tool_shed_repository.id"), index=True),
Column("repository_dependency_id", Integer, ForeignKey("repository_dependency.id"), index=True))
install_model.RepositoryDependency.table = Table("repository_dependency", metadata,
Column("id", Integer, primary_key=True),
Column("create_time", DateTime, default=now),
Column("update_time", DateTime, default=now, onupdate=now),
Column("tool_shed_repository_id", Integer, ForeignKey("tool_shed_repository.id"), index=True, nullable=False))
install_model.ToolDependency.table = Table("tool_dependency", metadata,
Column("id", Integer, primary_key=True),
Column("create_time", DateTime, default=now),
Column("update_time", DateTime, default=now, onupdate=now),
Column("tool_shed_repository_id", Integer, ForeignKey("tool_shed_repository.id"), index=True, nullable=False),
Column("name", TrimmedString(255)),
Column("version", TEXT),
Column("type", TrimmedString(40)),
Column("status", TrimmedString(255), nullable=False),
Column("error_message", TEXT))
install_model.ToolVersion.table = Table("tool_version", metadata,
Column("id", Integer, primary_key=True),
Column("create_time", DateTime, default=now),
Column("update_time", DateTime, default=now, onupdate=now),
Column("tool_id", String(255)),
Column("tool_shed_repository_id", Integer, ForeignKey("tool_shed_repository.id"), index=True, nullable=True))
install_model.ToolVersionAssociation.table = Table("tool_version_association", metadata,
Column("id", Integer, primary_key=True),
Column("tool_id", Integer, ForeignKey("tool_version.id"), index=True, nullable=False),
Column("parent_id", Integer, ForeignKey("tool_version.id"), index=True, nullable=False))
install_model.MigrateTools.table = Table("migrate_tools", metadata,
Column("repository_id", TrimmedString(255)),
Column("repository_path", TEXT),
Column("version", Integer))
mapper(install_model.ToolShedRepository, install_model.ToolShedRepository.table,
properties=dict(tool_versions=relation(install_model.ToolVersion,
primaryjoin=(install_model.ToolShedRepository.table.c.id == install_model.ToolVersion.table.c.tool_shed_repository_id),
backref='tool_shed_repository'),
tool_dependencies=relation(install_model.ToolDependency,
primaryjoin=(install_model.ToolShedRepository.table.c.id == install_model.ToolDependency.table.c.tool_shed_repository_id),
order_by=install_model.ToolDependency.table.c.name,
backref='tool_shed_repository'),
required_repositories=relation(install_model.RepositoryRepositoryDependencyAssociation,
primaryjoin=(install_model.ToolShedRepository.table.c.id == install_model.RepositoryRepositoryDependencyAssociation.table.c.tool_shed_repository_id))))
mapper(install_model.RepositoryRepositoryDependencyAssociation, install_model.RepositoryRepositoryDependencyAssociation.table,
properties=dict(repository=relation(install_model.ToolShedRepository,
primaryjoin=(install_model.RepositoryRepositoryDependencyAssociation.table.c.tool_shed_repository_id == install_model.ToolShedRepository.table.c.id)),
repository_dependency=relation(install_model.RepositoryDependency,
primaryjoin=(install_model.RepositoryRepositoryDependencyAssociation.table.c.repository_dependency_id == install_model.RepositoryDependency.table.c.id))))
mapper(install_model.RepositoryDependency, install_model.RepositoryDependency.table,
properties=dict(repository=relation(install_model.ToolShedRepository,
primaryjoin=(install_model.RepositoryDependency.table.c.tool_shed_repository_id == install_model.ToolShedRepository.table.c.id))))
mapper(install_model.ToolDependency, install_model.ToolDependency.table)
mapper(install_model.ToolVersion, install_model.ToolVersion.table,
properties=dict(
parent_tool_association=relation(install_model.ToolVersionAssociation,
primaryjoin=(install_model.ToolVersion.table.c.id == install_model.ToolVersionAssociation.table.c.tool_id)),
child_tool_association=relation(install_model.ToolVersionAssociation,
primaryjoin=(install_model.ToolVersion.table.c.id == install_model.ToolVersionAssociation.table.c.parent_id))))
mapper(install_model.ToolVersionAssociation, install_model.ToolVersionAssociation.table)
metadata = mapper_registry.metadata
def init(url, engine_options=None, create_tables=False):
lib/galaxy/tool_shed/galaxy_install/datatypes/custom_datatype_manager.py
+1 -1
View File
@@ -154,7 +154,7 @@ class CustomDatatypeLoader:
Load proprietary datatypes and return information needed for loading custom
datatypes converters and display applications later.
"""
metadata = repository.metadata
metadata = repository.metadata_
repository_dict = None
datatypes_config = get_config_from_disk(DATATYPES_CONFIG_FILENAME, relative_install_dir)
if datatypes_config:
lib/galaxy/tool_shed/galaxy_install/install_manager.py
+3 -3
View File
@@ -515,7 +515,7 @@ class InstallRepositoryManager:
persist=True)
irmm.generate_metadata_for_changeset_revision()
irmm_metadata_dict = irmm.get_metadata_dict()
tool_shed_repository.metadata = irmm_metadata_dict
tool_shed_repository.metadata_ = irmm_metadata_dict
# Update the tool_shed_repository.tool_shed_status column in the database.
tool_shed_status_dict = repository_util.get_tool_shed_status_for_installed_repository(self.app, tool_shed_repository)
if tool_shed_status_dict:
@@ -934,7 +934,7 @@ class InstallRepositoryManager:
shed_tool_conf=shed_tool_conf,
reinstalling=reinstalling,
tool_panel_section_mapping=tool_panel_section_mapping)
metadata = tool_shed_repository.metadata
metadata = tool_shed_repository.metadata_
if 'tools' in metadata and install_resolver_dependencies:
self.update_tool_shed_repository_status(tool_shed_repository,
self.install_model.ToolShedRepository.installation_status.INSTALLING_TOOL_DEPENDENCIES)
@@ -964,7 +964,7 @@ class InstallRepositoryManager:
def update_tool_shed_repository(self, repository, tool_shed_url, latest_ctx_rev, latest_changeset_revision,
install_new_dependencies=True, install_options=None):
install_options = install_options or {}
original_metadata_dict = repository.metadata
original_metadata_dict = repository.metadata_
original_repository_dependencies_dict = original_metadata_dict.get('repository_dependencies', {})
original_repository_dependencies = original_repository_dependencies_dict.get('repository_dependencies', [])
original_tool_dependencies_dict = original_metadata_dict.get('tool_dependencies', {})
lib/galaxy/tool_shed/galaxy_install/installed_repository_manager.py
+5 -5
View File
@@ -87,10 +87,10 @@ class InstalledRepositoryManager:
tpm=tpm,
repository=repository,
changeset_revision=repository.changeset_revision,
metadata_dict=repository.metadata)
metadata_dict=repository.metadata_)
repository_tools_tups = irmm.get_repository_tools_tups()
# Reload tools into the appropriate tool panel section.
tool_panel_dict = repository.metadata['tool_panel_section']
tool_panel_dict = repository.metadata_['tool_panel_section']
tpm.add_to_tool_panel(repository.name,
repository_clone_url,
repository.installed_changeset_revision,
@@ -105,7 +105,7 @@ class InstalledRepositoryManager:
data_manager_relative_install_dir = os.path.join(data_manager_relative_install_dir, repository.name)
dmh = data_manager.DataManagerHandler(self.app)
dmh.install_data_managers(self.app.config.shed_data_manager_config_file,
repository.metadata,
repository.metadata_,
repository.get_shed_config_dict(self.app),
data_manager_relative_install_dir,
repository,
@@ -204,7 +204,7 @@ class InstalledRepositoryManager:
name,
repository_owner,
changeset_revision)
if not updating and repository and repository.metadata:
if not updating and repository and repository.metadata_:
installed_rd, missing_rd = self.get_installed_and_missing_repository_dependencies(repository)
else:
installed_rd, missing_rd = \
@@ -286,7 +286,7 @@ class InstalledRepositoryManager:
if has_repository_dependencies:
# The repository dependencies container will include only the immediate repository
# dependencies of this repository, so the container will be only a single level in depth.
metadata = repository.metadata
metadata = repository.metadata_
installed_rd_tups = []
missing_rd_tups = []
for tsr in repository.repository_dependencies:
lib/galaxy/tool_shed/galaxy_install/metadata/installed_repository_metadata_manager.py
+2 -2
View File
@@ -87,10 +87,10 @@ class InstalledRepositoryMetadataManager(MetadataGenerator):
def reset_all_metadata_on_installed_repository(self):
"""Reset all metadata on a single tool shed repository installed into a Galaxy instance."""
if self.relative_install_dir:
original_metadata_dict = self.repository.metadata
original_metadata_dict = self.repository.metadata_
self.generate_metadata_for_changeset_revision()
if self.metadata_dict != original_metadata_dict:
self.repository.metadata = self.metadata_dict
self.repository.metadata_ = self.metadata_dict
self.update_in_shed_tool_config()
self.app.install_model.context.add(self.repository)
self.app.install_model.context.flush()
lib/galaxy/tool_shed/galaxy_install/repository_dependencies/repository_dependency_manager.py
+2 -2
View File
@@ -233,10 +233,10 @@ class RepositoryDependencyInstallManager:
# The database record for the tool shed repository currently being processed can be updated.
# Get the repository metadata to see where it was previously located in the tool panel.
tpm = tool_panel_manager.ToolPanelManager(self.app)
if repository_db_record and repository_db_record.metadata:
if repository_db_record and repository_db_record.metadata_:
_, tool_panel_section_key = \
tpm.handle_tool_panel_selection(toolbox=self.app.toolbox,
metadata=repository_db_record.metadata,
metadata=repository_db_record.metadata_,
no_changes_checked=no_changes_checked,
tool_panel_section_id=tool_panel_section_id,
new_tool_panel_section_label=new_tool_panel_section_label)
lib/galaxy/tool_shed/galaxy_install/tool_migration_manager.py
+1 -1
View File
@@ -442,7 +442,7 @@ class ToolMigrationManager:
persist=True)
irmm.generate_metadata_for_changeset_revision()
irmm_metadata_dict = irmm.get_metadata_dict()
tool_shed_repository.metadata = irmm_metadata_dict
tool_shed_repository.metadata_ = irmm_metadata_dict
self.app.install_model.context.add(tool_shed_repository)
self.app.install_model.context.flush()
has_tool_dependencies = self.__has_tool_dependencies(irmm_metadata_dict)
lib/galaxy/tool_shed/galaxy_install/tools/data_manager.py
+1 -1
View File
@@ -149,7 +149,7 @@ class DataManagerHandler:
return rval
def remove_from_data_manager(self, repository):
metadata_dict = repository.metadata
metadata_dict = repository.metadata_
if metadata_dict and 'data_manager' in metadata_dict:
shed_data_manager_conf_filename = self.app.config.shed_data_manager_config_file
tree, error_message = parse_xml(shed_data_manager_conf_filename)
lib/galaxy/tool_shed/galaxy_install/tools/tool_panel_manager.py
+3 -3
View File
@@ -212,7 +212,7 @@ class ToolPanelManager:
tool_panel_dict = {}
shed_tool_conf, tool_path, relative_install_dir = \
get_tool_panel_config_tool_path_install_dir(self.app, repository)
metadata = repository.metadata
metadata = repository.metadata_
# Create a dictionary of tool guid and tool config file name for each tool in the repository.
guids_and_configs = {}
if 'tools' in metadata:
@@ -467,7 +467,7 @@ class ToolPanelManager:
# information so the tools can be displayed in the same way when the repository is
# activated or reinstalled.
tool_panel_dict = self.generate_tool_panel_dict_from_shed_tool_conf_entries(repository)
repository.metadata['tool_panel_section'] = tool_panel_dict
repository.metadata_['tool_panel_section'] = tool_panel_dict
self.app.install_model.context.add(repository)
self.app.install_model.context.flush()
# Create a list of guids for all tools that will be removed from the in-memory tool panel
@@ -481,7 +481,7 @@ class ToolPanelManager:
shed_tool_conf_dict = self.get_shed_tool_conf_dict(shed_tool_conf)
if uninstall:
# Remove from the shed_tool_conf file on disk.
self.remove_from_shed_tool_config(shed_tool_conf_dict, repository.metadata)
self.remove_from_shed_tool_config(shed_tool_conf_dict, repository.metadata_)
def update_tool_panel_dict(self, tool_panel_dict, tool_panel_section_mapping, repository_tools_tups):
for tool_guid in tool_panel_dict:
lib/galaxy/tool_shed/galaxy_install/update_repository_manager.py
+1 -1
View File
@@ -109,7 +109,7 @@ class UpdateRepositoryManager:
Update a tool_shed_repository database record with new information retrieved from the
Tool Shed. This happens when updating an installed repository to a new changeset revision.
"""
repository.metadata = updated_metadata_dict
repository.metadata_ = updated_metadata_dict
tool_shed_url = get_tool_shed_url_from_tool_shed_registry(self.app, repository.tool_shed)
clean_dependency_relationships(self.app, updated_metadata_dict, repository, tool_shed_url)
# Update the repository.changeset_revision column in the database.
lib/galaxy/tool_shed/metadata/metadata_generator.py
+1 -1
View File
@@ -317,7 +317,7 @@ class MetadataGenerator:
if self.updating_installed_repository:
# Keep the original tool shed repository metadata if setting metadata on a repository
# installed into a local Galaxy instance for which we have pulled updates.
original_repository_metadata = self.repository.metadata
original_repository_metadata = self.repository.metadata_
else:
original_repository_metadata = None
readme_file_names = _get_readme_file_names(str(self.repository.name))
lib/galaxy/tool_shed/util/repository_util.py
+2 -2
View File
@@ -112,7 +112,7 @@ def create_or_update_tool_shed_repository(app, name, description, installed_chan
tool_shed_repository.description = description
tool_shed_repository.changeset_revision = current_changeset_revision
tool_shed_repository.ctx_rev = ctx_rev
tool_shed_repository.metadata = metadata_dict
tool_shed_repository.metadata_ = metadata_dict
tool_shed_repository.includes_datatypes = includes_datatypes
tool_shed_repository.deleted = deleted
tool_shed_repository.uninstalled = uninstalled
@@ -127,7 +127,7 @@ def create_or_update_tool_shed_repository(app, name, description, installed_chan
installed_changeset_revision=installed_changeset_revision,
changeset_revision=current_changeset_revision,
ctx_rev=ctx_rev,
metadata=metadata_dict,
metadata_=metadata_dict,
includes_datatypes=includes_datatypes,
dist_to_shed=dist_to_shed,
deleted=deleted,
lib/galaxy/tools/cache.py
+1 -1
View File
@@ -301,7 +301,7 @@ class ToolShedRepositoryCache:
def rebuild(self):
self.repositories = self.session.query(ToolShedRepository).options(
defer(ToolShedRepository.metadata), joinedload('tool_dependencies')
defer(ToolShedRepository.metadata_), joinedload('tool_dependencies')
).all()
repos_by_tuple = defaultdict(list)
for repository in self.repositories + self.local_repositories:
lib/galaxy/webapps/galaxy/controllers/admin_toolshed.py
+3 -3
View File
@@ -177,7 +177,7 @@ class AdminToolshed(AdminGalaxy):
message = escape(kwd.get('message', ''))
status = kwd.get('status', 'done')
repository = repository_util.get_installed_tool_shed_repository(trans.app, repository_id)
repository_metadata = repository.metadata
repository_metadata = repository.metadata_
shed_config_dict = repository.get_shed_config_dict(trans.app)
tool_metadata = {}
tool_lineage = []
@@ -863,7 +863,7 @@ class AdminToolshed(AdminGalaxy):
new_tool_panel_section_label = kwd.get('new_tool_panel_section_label', '')
tool_panel_section_key = None
tool_panel_section_keys = []
metadata = tool_shed_repository.metadata
metadata = tool_shed_repository.metadata_
# Keep track of tool dependencies defined for the current repository or those defined for any of
# its repository dependencies.
includes_tool_dependencies = tool_shed_repository.includes_tool_dependencies
@@ -1024,7 +1024,7 @@ class AdminToolshed(AdminGalaxy):
latest_ctx_rev = kwd.get('latest_ctx_rev', None)
tool_shed_repository = repository_util.get_installed_tool_shed_repository(trans.app, repository_id)
repository_clone_url = common_util.generate_clone_url_for_installed_repository(trans.app, tool_shed_repository)
metadata = tool_shed_repository.metadata
metadata = tool_shed_repository.metadata_
tool_shed_url = common_util.get_tool_shed_url_from_tool_shed_registry(trans.app, str(tool_shed_repository.tool_shed))
tool_path = tool_shed_repository.get_tool_relative_path(trans.app)[0]
if latest_changeset_revision and latest_ctx_rev:
lib/tool_shed/galaxy_install/dependency_display.py
+2 -2
View File
@@ -441,7 +441,7 @@ class DependencyDisplayer:
when displaying repository dependencies for installed repositories and when displaying
them for uninstalled repositories that are being reinstalled.
"""
metadata = repository.metadata
metadata = repository.metadata_
if metadata:
# Handle proprietary datatypes.
datatypes = metadata.get('datatypes', None)
@@ -464,7 +464,7 @@ class DependencyDisplayer:
readme_files_dict = readme_util.build_readme_files_dict(self.app,
repository,
repository.changeset_revision,
repository.metadata, tool_path)
repository.metadata_, tool_path)
else:
readme_files_dict = None
# Handle repository dependencies.
lib/tool_shed/test/base/twilltestcase.py
+6 -6
View File
@@ -321,7 +321,7 @@ class ShedTwillTestCase(DrivenFunctionalTestCase):
(installed_repository.status, expected_status)
def check_galaxy_repository_tool_panel_section(self, repository, expected_tool_panel_section):
metadata = repository.metadata
metadata = repository.metadata_
assert 'tools' in metadata, f'Tools not found in repository metadata: {metadata}'
# If integrated_tool_panel.xml is to be tested, this test method will need to be enhanced to handle tools
# from the same repository in different tool panel sections. Getting the first tool guid is ok, because
@@ -571,7 +571,7 @@ class ShedTwillTestCase(DrivenFunctionalTestCase):
self.check_for_strings(strings_displayed, strings_not_displayed)
def display_installed_jobs_list_page(self, installed_repository, data_manager_names=None, strings_displayed=None, strings_not_displayed=None):
data_managers = installed_repository.metadata.get('data_manager', {}).get('data_managers', {})
data_managers = installed_repository.metadata_.get('data_manager', {}).get('data_managers', {})
if data_manager_names:
if not isinstance(data_manager_names, list):
data_manager_names = [data_manager_names]
@@ -1500,14 +1500,14 @@ class ShedTwillTestCase(DrivenFunctionalTestCase):
def verify_installed_repository_metadata_unchanged(self, name, owner):
installed_repository = test_db_util.get_installed_repository_by_name_owner(name, owner)
metadata = installed_repository.metadata
metadata = installed_repository.metadata_
self.reset_installed_repository_metadata(installed_repository)
new_metadata = installed_repository.metadata
new_metadata = installed_repository.metadata_
assert metadata == new_metadata, f'Metadata for installed repository {name} differs after metadata reset.'
def verify_installed_repository_no_tool_panel_section(self, repository):
'''Verify that there is no 'tool_panel_section' entry in the repository metadata.'''
metadata = repository.metadata
metadata = repository.metadata_
assert 'tool_panel_section' not in metadata, f'Tool panel section incorrectly found in metadata: {metadata}'
def verify_installed_repository_data_table_entries(self, required_data_table_entries):
@@ -1582,7 +1582,7 @@ class ShedTwillTestCase(DrivenFunctionalTestCase):
if strings_not_displayed is None:
strings_not_displayed = []
repository_id = self.security.encode_id(installed_repository.id)
for tool in installed_repository.metadata['tools']:
for tool in installed_repository.metadata_['tools']:
strings = list(strings_displayed)
strings.extend([tool['id'], tool['description'], tool['version'], tool['guid'], tool['name']])
params = dict(repository_id=repository_id, tool_id=tool['id'])
lib/tool_shed/test/functional/test_1070_invalid_tool.py
+1 -1
View File
@@ -84,5 +84,5 @@ class TestFreebayesRepository(ShedTwillTestCase):
strings_not_displayed=['bisulfite mapper'])
self.verify_tool_metadata_for_installed_repository(installed_repository)
self.update_installed_repository(installed_repository, strings_displayed=["there are no updates available"])
assert 'invalid_tools' in installed_repository.metadata, 'No invalid tools were defined in %s.' % \
assert 'invalid_tools' in installed_repository.metadata_, 'No invalid tools were defined in %s.' % \
installed_repository.name
lib/tool_shed/test/functional/test_1300_reset_all_metadata.py
+4 -4
View File
@@ -441,7 +441,7 @@ class TestResetInstalledRepositoryMetadata(ShedTwillTestCase):
repository_metadata = dict()
repositories = self.test_db_util.get_all_installed_repositories(actually_installed=True)
for repository in repositories:
repository_metadata[self.security.encode_id(repository.id)] = repository.metadata
repository_metadata[self.security.encode_id(repository.id)] = repository.metadata_
self.reset_metadata_on_selected_installed_repositories(list(repository_metadata.keys()))
for repository in repositories:
self.test_db_util.ga_refresh(repository)
@@ -452,7 +452,7 @@ class TestResetInstalledRepositoryMetadata(ShedTwillTestCase):
# is normal and expected behavior, the functional tests assume that repository metadata will not change
# in any way after a reset. A workaround is to remove the tool panel section from the stored repository
# metadata dict, in order to eliminate the misleading detection of changed metadata.
if 'tool_panel_section' in old_metadata and 'tool_panel_section' not in repository.metadata:
if 'tool_panel_section' in old_metadata and 'tool_panel_section' not in repository.metadata_:
del old_metadata['tool_panel_section']
assert repository.metadata == old_metadata, 'Metadata for %s repository %s changed after reset. \nOld: %s\nNew: %s' % \
(repository.status, repository.name, old_metadata, repository.metadata)
assert repository.metadata_ == old_metadata, 'Metadata for %s repository %s changed after reset. \nOld: %s\nNew: %s' % \
(repository.status, repository.name, old_metadata, repository.metadata_)
lib/tool_shed/util/shed_util_common.py
+1 -1
View File
@@ -175,7 +175,7 @@ def get_requirements_from_repository(repository):
if not repository.includes_tools:
return {}
else:
return get_requirements_from_tools(repository.metadata.get('tools', []))
return get_requirements_from_tools(repository.metadata_.get('tools', []))
def get_repository_categories(app, id):
scripts/update_shed_config_path.py
+3 -3
View File
@@ -16,9 +16,9 @@ def main(opts, session, model):
Find all tool shed repositories with the bad path and update with the correct path.
'''
for row in session.query(model.ToolShedRepository).all():
if 'shed_config_filename' in row.metadata:
if row.metadata['shed_config_filename'] == opts.bad_filename:
row.metadata['shed_config_filename'] = opts.good_filename
if 'shed_config_filename' in row.metadata_:
if row.metadata_['shed_config_filename'] == opts.bad_filename:
row.metadata_['shed_config_filename'] = opts.good_filename
session.add(row)
session.flush()
return 0
test/unit/app/tools/conftest.py
+1 -1
View File
@@ -56,7 +56,7 @@ def create_repo(session, changeset, installed_changeset, config_filename=None):
}
if config_filename:
metadata['shed_config_filename'] = config_filename
repository = tool_shed_install.ToolShedRepository(metadata=metadata)
repository = tool_shed_install.ToolShedRepository(metadata_=metadata)
repository.tool_shed = "github.com"
repository.owner = "galaxyproject"
repository.name = "example"
test/unit/app/tools/test_toolbox.py
+1 -1
View File
@@ -111,7 +111,7 @@ class BaseToolBoxTestCase(unittest.TestCase, UsesTools):
}
if config_filename:
metadata['shed_config_filename'] = config_filename
repository = tool_shed_install.ToolShedRepository(metadata=metadata)
repository = tool_shed_install.ToolShedRepository(metadata_=metadata)
repository.tool_shed = DEFAULT_TEST_REPO.tool_shed
repository.owner = DEFAULT_TEST_REPO.owner
repository.name = DEFAULT_TEST_REPO.name
test/unit/data/model/test_install_model_mapping.py
+431
View File
@@ -0,0 +1,431 @@
from contextlib import contextmanager
from datetime import datetime, timedelta
import pytest
from sqlalchemy import (
delete,
select,
)
import galaxy.model.tool_shed_install.mapping as mapping
class BaseTest:
@pytest.fixture
def cls_(self, model):
"""
Return class under test.
Assumptions: if the class under test is Foo, then the class grouping
the tests should be a subclass of BaseTest, named TestFoo.
"""
prefix = len('Test')
class_name = self.__class__.__name__[prefix:]
return getattr(model, class_name)
class TestToolShedRepository(BaseTest):
def test_table(self, cls_):
assert cls_.__tablename__ == 'tool_shed_repository'
def test_columns(self, session, cls_, repository):
create_time = datetime.now()
update_time = create_time + timedelta(hours=1)
tool_shed = 'a'
name = 'b'
description = 'c'
owner = 'd'
installed_changeset_revision = 'e'
changeset_revision = 'f'
ctx_rev = 'g'
metadata_ = 'h'
includes_datatypes = True
tool_shed_status = 'i'
deleted = True
uninstalled = True
dist_to_shed = True
status = 'j'
error_message = 'k'
obj = cls_()
obj.create_time = create_time
obj.update_time = update_time
obj.tool_shed = tool_shed
obj.name = name
obj.description = description
obj.owner = owner
obj.installed_changeset_revision = installed_changeset_revision
obj.changeset_revision = changeset_revision
obj.ctx_rev = ctx_rev
obj.metadata_ = metadata_
obj.includes_datatypes = includes_datatypes
obj.tool_shed_status = tool_shed_status
obj.deleted = deleted
obj.uninstalled = uninstalled
obj.dist_to_shed = dist_to_shed
obj.status = status
obj.error_message = error_message
with dbcleanup(session, obj) as obj_id:
stored_obj = get_stored_obj(session, cls_, obj_id)
assert stored_obj.id == obj_id
assert stored_obj.create_time == create_time
assert stored_obj.update_time == update_time
assert stored_obj.tool_shed == tool_shed
assert stored_obj.name == name
assert stored_obj.description == description
assert stored_obj.owner == owner
assert stored_obj.installed_changeset_revision == installed_changeset_revision
assert stored_obj.changeset_revision == changeset_revision
assert stored_obj.ctx_rev == ctx_rev
assert stored_obj.metadata_ == metadata_
assert stored_obj.includes_datatypes == includes_datatypes
assert stored_obj.tool_shed_status == tool_shed_status
assert stored_obj.deleted == deleted
assert stored_obj.uninstalled == uninstalled
assert stored_obj.dist_to_shed == dist_to_shed
assert stored_obj.status == status
assert stored_obj.error_message == error_message
def test_relationships(
self,
session, cls_,
repository,
tool_version,
tool_dependency,
repository_repository_dependency_association,
):
obj = cls_()
obj.tool_versions.append(tool_version)
obj.tool_dependencies.append(tool_dependency)
obj.required_repositories.append(repository_repository_dependency_association)
with dbcleanup(session, obj) as obj_id:
stored_obj = get_stored_obj(session, cls_, obj_id)
assert stored_obj.tool_versions == [tool_version]
assert stored_obj.tool_dependencies == [tool_dependency]
assert stored_obj.required_repositories == [repository_repository_dependency_association]
class TestRepositoryRepositoryDependencyAssociation(BaseTest):
def test_table(self, cls_):
assert cls_.__tablename__ == 'repository_repository_dependency_association'
def test_columns(self, session, cls_, repository, repository_dependency):
create_time = datetime.now()
update_time = create_time + timedelta(hours=1)
obj = cls_()
obj.create_time = create_time
obj.update_time = update_time
obj.repository = repository
obj.repository_dependency = repository_dependency
with dbcleanup(session, obj) as obj_id:
stored_obj = get_stored_obj(session, cls_, obj_id)
assert stored_obj.id == obj_id
assert stored_obj.create_time == create_time
assert stored_obj.update_time == update_time
assert stored_obj.tool_shed_repository_id == repository.id
assert stored_obj.repository_dependency_id == repository_dependency.id
def test_relationships(self, session, cls_, repository, repository_dependency):
obj = cls_()
obj.repository = repository
obj.repository_dependency = repository_dependency
with dbcleanup(session, obj) as obj_id:
stored_obj = get_stored_obj(session, cls_, obj_id)
assert stored_obj.repository.id == repository.id
assert stored_obj.repository_dependency.id == repository_dependency.id
class TestRepositoryDependency(BaseTest):
def test_table(self, cls_):
assert cls_.__tablename__ == 'repository_dependency'
def test_columns(self, session, cls_, repository):
create_time = datetime.now()
update_time = create_time + timedelta(hours=1)
obj = cls_()
obj.create_time = create_time
obj.update_time = update_time
obj.repository = repository
with dbcleanup(session, obj) as obj_id:
stored_obj = get_stored_obj(session, cls_, obj_id)
assert stored_obj.id == obj_id
assert stored_obj.create_time == create_time
assert stored_obj.update_time == update_time
assert stored_obj.tool_shed_repository_id == repository.id
def test_relationships(self, session, cls_, repository):
obj = cls_()
obj.repository = repository
with dbcleanup(session, obj) as obj_id:
stored_obj = get_stored_obj(session, cls_, obj_id)
assert stored_obj.repository.id == repository.id
class TestToolDependency(BaseTest):
def test_table(self, cls_):
assert cls_.__tablename__ == 'tool_dependency'
def test_columns(self, session, cls_, repository):
create_time = datetime.now()
update_time = create_time + timedelta(hours=1)
name, version, type, status, error_message = 'a', 'b', 'c', 'd', 'e'
obj = cls_()
obj.create_time = create_time
obj.update_time = update_time
obj.tool_shed_repository_id = repository.id
obj.name = name
obj.version = version
obj.type = type
obj.status = status
obj.error_message = error_message
with dbcleanup(session, obj) as obj_id:
stored_obj = get_stored_obj(session, cls_, obj_id)
assert stored_obj.id == obj_id
assert stored_obj.create_time == create_time
assert stored_obj.update_time == update_time
assert stored_obj.tool_shed_repository_id == repository.id
assert stored_obj.name == name
assert stored_obj.version == version
assert stored_obj.type == type
assert stored_obj.status == status
assert stored_obj.error_message == error_message
def test_relationships(self, session, cls_, repository):
obj = cls_()
obj.tool_shed_repository = repository
obj.status = 'a'
with dbcleanup(session, obj) as obj_id:
stored_obj = get_stored_obj(session, cls_, obj_id)
assert stored_obj.tool_shed_repository.id == repository.id
class TestToolVersion(BaseTest):
def test_table(self, cls_):
assert cls_.__tablename__ == 'tool_version'
def test_columns(self, session, cls_, repository):
create_time = datetime.now()
update_time = create_time + timedelta(hours=1)
tool_id = 'a'
obj = cls_()
obj.create_time = create_time
obj.update_time = update_time
obj.tool_id = tool_id
obj.tool_shed_repository = repository
with dbcleanup(session, obj) as obj_id:
stored_obj = get_stored_obj(session, cls_, obj_id)
assert stored_obj.id == obj_id
assert stored_obj.create_time == create_time
assert stored_obj.update_time == update_time
assert stored_obj.tool_id == tool_id
assert stored_obj.tool_shed_repository_id == repository.id
def test_relationships(self, session, cls_, repository, tool_version_association_factory, tool_version):
# This test is non-standard because we must test associations that do not have relationships set up.
# As a result, we need the object pkey in order to setup the test, which is why we have to manually
# add obj to the session and flush it.
obj = cls_()
obj.tool_shed_repository = repository
session.add(obj)
session.flush()
tool_version_assoc1 = tool_version_association_factory()
tool_version_assoc1.tool_id = obj.id # tool_version under test
tool_version_assoc1.parent_id = tool_version.id # some other tool_version
tool_version_assoc2 = tool_version_association_factory()
tool_version_assoc2.tool_id = tool_version.id # some other tool_version
tool_version_assoc2.parent_id = obj.id # tool_version under test
session.add(tool_version_assoc1)
session.add(tool_version_assoc2)
session.flush()
stored_obj = get_stored_obj(session, cls_, obj.id)
assert stored_obj.tool_shed_repository.id == repository.id
assert stored_obj.parent_tool_association == [tool_version_assoc1]
assert stored_obj.child_tool_association == [tool_version_assoc2]
delete_from_database(session, [obj, tool_version_assoc1, tool_version_assoc2])
class TestToolVersionAssociation(BaseTest):
def test_table(self, cls_):
assert cls_.__tablename__ == 'tool_version_association'
def test_columns(self, session, cls_, tool_version_factory):
tool_version = tool_version_factory()
parent_tool_version = tool_version_factory()
session.add(tool_version)
session.add(parent_tool_version)
session.flush()
# TODO: why are these not mapped as relationships?
obj = cls_()
obj.tool_id = tool_version.id
obj.parent_id = parent_tool_version.id
with dbcleanup(session, obj) as obj_id:
stored_obj = get_stored_obj(session, cls_, obj_id)
assert stored_obj.id == obj_id
assert stored_obj.tool_id == tool_version.id
assert stored_obj.parent_id == parent_tool_version.id
delete_from_database(session, [tool_version, parent_tool_version])
# Misc. helper fixtures.
@pytest.fixture(scope='module')
def model():
db_uri = 'sqlite:///:memory:'
return mapping.init(db_uri, create_tables=True)
@pytest.fixture
def session(model):
Session = model.session
yield Session()
Session.remove() # Ensures we get a new session for each test
# Fixtures yielding persisted instances of models, deleted from the database on test exit.
@pytest.fixture
def repository(model, session):
instance = model.ToolShedRepository()
yield from dbcleanup_wrapper(session, instance)
@pytest.fixture
def repository_repository_dependency_association(model, session):
instance = model.RepositoryRepositoryDependencyAssociation()
yield from dbcleanup_wrapper(session, instance)
@pytest.fixture
def repository_dependency(model, session, repository):
instance = model.RepositoryDependency()
instance.repository = repository
yield from dbcleanup_wrapper(session, instance)
@pytest.fixture
def tool_dependency(model, session, repository):
instance = model.ToolDependency()
instance.tool_shed_repository = repository
instance.status = 'a'
yield from dbcleanup_wrapper(session, instance)
@pytest.fixture
def tool_version(model, session):
instance = model.ToolVersion()
yield from dbcleanup_wrapper(session, instance)
@pytest.fixture
def tool_version_association_factory(model):
def make_instance(*args, **kwds):
return model.ToolVersionAssociation(*args, **kwds)
return make_instance
@pytest.fixture
def tool_version_factory(model):
def make_instance(*args, **kwds):
return model.ToolVersion(*args, **kwds)
return make_instance
# Test utilities
def dbcleanup_wrapper(session, obj, where_clause=None):
with dbcleanup(session, obj, where_clause):
yield obj
@contextmanager
def dbcleanup(session, obj, where_clause=None):
"""
Use the session to store obj in database; delete from database on exit, bypassing the session.
If obj does not have an id field, a SQLAlchemy WHERE clause should be provided to construct
a custom select statement.
"""
return_id = where_clause is None
try:
obj_id = persist(session, obj, return_id)
yield obj_id
finally:
# table = obj.__table__ # TODO
table = obj.table
if where_clause is None:
where_clause = _get_default_where_clause(type(obj), obj_id)
stmt = delete(table).where(where_clause)
session.execute(stmt)
def persist(session, obj, return_id=True):
"""
Use the session to store obj in database, then remove obj from session,
so that on a subsequent load from the database we get a clean instance.
"""
session.add(obj)
session.flush()
obj_id = obj.id if return_id else None # save this before obj is expunged
session.expunge(obj)
return obj_id
def get_stored_obj(session, cls, obj_id=None, where_clause=None, unique=False):
# Either obj_id or where_clause must be provided, but not both
assert bool(obj_id) ^ (where_clause is not None)
if where_clause is None:
where_clause = _get_default_where_clause(cls, obj_id)
stmt = select(cls).where(where_clause)
result = session.execute(stmt)
# unique() is required if result contains joint eager loads against collections
# https://gerrit.sqlalchemy.org/c/sqlalchemy/sqlalchemy/+/2253
if unique:
result = result.unique()
return result.scalar_one()
def _get_default_where_clause(cls, obj_id):
# where_clause = cls.__table__.c.id == obj_id # TODO
where_clause = cls.table.c.id == obj_id
return where_clause
def delete_from_database(session, objects):
"""
Delete each object in objects from database.
May be called at the end of a test if use of a context manager is impractical.
(Assume all objects have the id field as their primary key.)
"""
# Ensure we have a list of objects (check for list explicitly: a model can be iterable)
if not isinstance(objects, list):
objects = [objects]
for obj in objects:
table = obj.__table__
stmt = delete(table).where(table.c.id == obj.id)
session.execute(stmt)
test/unit/data/model/test_mapping.py → test/unit/data/model/test_model_mapping.py
View File
test/unit/data/model/test_test_mapping.py
+1 -1
View File
@@ -17,7 +17,7 @@ from sqlalchemy.exc import NoResultFound
from sqlalchemy.orm import registry, Session
from galaxy.model import _HasTable
from . test_mapping import (
from . test_model_mapping import (
are_same_entity_collections,
dbcleanup,
dbcleanup_wrapper,