Warning
This document is for an in-development version of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.
Source code for tool_shed.webapp.model
import logging
import os
import weakref
from datetime import (
datetime,
timedelta
)
from typing import Any, Mapping, TYPE_CHECKING
from sqlalchemy import (
Boolean,
Column,
DateTime,
desc,
false,
ForeignKey,
Integer,
not_,
String,
Table,
TEXT,
true,
UniqueConstraint,
)
from sqlalchemy.orm import (
registry,
relationship,
)
from sqlalchemy.orm.decl_api import DeclarativeMeta
import tool_shed.repository_types.util as rt_util
from galaxy import util
from galaxy.model.custom_types import MutableJSONType, TrimmedString
from galaxy.model.orm.now import now
from galaxy.security.validate_user_input import validate_password_str
from galaxy.util import unique_id
from galaxy.util.bunch import Bunch
from galaxy.util.dictifiable import Dictifiable
from galaxy.util.hash_util import new_secure_hash
from tool_shed.dependencies.repository import relation_builder
from tool_shed.util import (
hg_util,
metadata_util
)
from tool_shed.util.hgweb_config import hgweb_config_manager
log = logging.getLogger(__name__)
WEAK_HG_REPO_CACHE: Mapping['Repository', Any] = weakref.WeakKeyDictionary()
if TYPE_CHECKING:
class _HasTable:
table: Table
else:
_HasTable = object
mapper_registry = registry()
[docs]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__
[docs]class APIKeys(Base, _HasTable):
__tablename__ = 'api_keys'
id = Column(Integer, primary_key=True)
create_time = Column(DateTime, default=now)
user_id = Column(ForeignKey('galaxy_user.id'), index=True)
key = Column(TrimmedString(32), index=True, unique=True)
user = relationship('User', back_populates='api_keys')
[docs]class User(Base, Dictifiable, _HasTable):
__tablename__ = 'galaxy_user'
id = Column(Integer, primary_key=True)
create_time = Column(DateTime, default=now)
update_time = Column(DateTime, default=now, onupdate=now)
email = Column(TrimmedString(255), nullable=False)
username = Column(String(255), index=True)
password = Column(TrimmedString(40), nullable=False)
external = Column(Boolean, default=False)
new_repo_alert = Column(Boolean, default=False)
deleted = Column(Boolean, index=True, default=False)
purged = Column(Boolean, index=True, default=False)
active_repositories = relationship('Repository',
primaryjoin=(lambda: (Repository.user_id == User.id) & (not_(Repository.deleted))), # type: ignore[has-type]
back_populates='user',
order_by=lambda: desc(Repository.name)) # type: ignore[has-type]
galaxy_sessions = relationship('GalaxySession',
back_populates='user',
order_by=lambda: desc(GalaxySession.update_time)) # type: ignore[has-type]
api_keys = relationship(
'APIKeys',
back_populates='user',
order_by=lambda: desc(APIKeys.create_time))
reset_tokens = relationship('PasswordResetToken', back_populates='user')
groups = relationship('UserGroupAssociation', back_populates='user')
dict_collection_visible_keys = ['id', 'username']
dict_element_visible_keys = ['id', 'username']
bootstrap_admin_user = False
roles = relationship('UserRoleAssociation', back_populates='user')
non_private_roles = relationship(
'UserRoleAssociation',
viewonly=True,
primaryjoin=(lambda:
(User.id == UserRoleAssociation.user_id) # type: ignore[has-type]
& (UserRoleAssociation.role_id == Role.id) # type: ignore[has-type]
& not_(Role.name == User.email)) # type: ignore[has-type]
)
repository_reviews = relationship('RepositoryReview', back_populates='user')
def __init__(self, email=None, password=None):
self.email = email
self.password = password
self.external = False
self.deleted = False
self.purged = False
self.new_repo_alert = False
[docs] def all_roles(self):
roles = [ura.role for ura in self.roles]
for group in [uga.group for uga in self.groups]:
for role in [gra.role for gra in group.roles]:
if role not in roles:
roles.append(role)
return roles
[docs] def check_password(self, cleartext):
"""Check if 'cleartext' matches 'self.password' when hashed."""
return self.password == new_secure_hash(text_type=cleartext)
@property
def nice_total_disk_usage(self):
return 0
total_disk_usage = property(get_disk_usage, set_disk_usage)
[docs] def set_password_cleartext(self, cleartext):
message = validate_password_str(cleartext)
if message:
raise Exception(f"Invalid password: {message}")
# Set 'self.password' to the digest of 'cleartext'.
self.password = new_secure_hash(text_type=cleartext)
[docs]class PasswordResetToken(Base, _HasTable):
__tablename__ = 'password_reset_token'
token = Column(String(32), primary_key=True, unique=True, index=True)
expiration_time = Column(DateTime)
user_id = Column(ForeignKey('galaxy_user.id'), index=True)
user = relationship('User', back_populates='reset_tokens')
def __init__(self, user, token=None):
if token:
self.token = token
else:
self.token = unique_id()
self.user = user
self.expiration_time = now() + timedelta(hours=24)
[docs]class Group(Base, Dictifiable, _HasTable):
__tablename__ = 'galaxy_group'
id = Column(Integer, primary_key=True)
create_time = Column(DateTime, default=now)
update_time = Column(DateTime, default=now, onupdate=now)
name = Column(String(255), index=True, unique=True)
deleted = Column(Boolean, index=True, default=False)
roles = relationship('GroupRoleAssociation', back_populates='group')
users = relationship('UserGroupAssociation', back_populates='group')
dict_collection_visible_keys = ['id', 'name']
dict_element_visible_keys = ['id', 'name']
def __init__(self, name=None):
self.name = name
self.deleted = False
[docs]class Role(Base, Dictifiable, _HasTable):
__tablename__ = 'role'
id = Column(Integer, primary_key=True)
create_time = Column(DateTime, default=now)
update_time = Column(DateTime, default=now, onupdate=now)
name = Column(String(255), index=True, unique=True)
description = Column(TEXT)
type = Column(String(40), index=True)
deleted = Column(Boolean, index=True, default=False)
repositories = relationship('RepositoryRoleAssociation', back_populates='role')
groups = relationship('GroupRoleAssociation', back_populates='role')
users = relationship('UserRoleAssociation', back_populates='role')
dict_collection_visible_keys = ['id', 'name']
dict_element_visible_keys = ['id', 'name', 'description', 'type']
private_id = None
types = Bunch(PRIVATE='private',
SYSTEM='system',
USER='user',
ADMIN='admin',
SHARING='sharing')
def __init__(self, name=None, description=None, type=types.SYSTEM, deleted=False):
self.name = name
self.description = description
self.type = type
self.deleted = deleted
@property
def is_repository_admin_role(self):
# A repository admin role must always be associated with a repository. The mapper returns an
# empty list for those roles that have no repositories. This method will require changes if
# new features are introduced that results in more than one role per repository.
if self.repositories:
return True
return False
[docs]class UserGroupAssociation(Base, _HasTable):
__tablename__ = 'user_group_association'
id = Column(Integer, primary_key=True)
user_id = Column(ForeignKey('galaxy_user.id'), index=True)
group_id = Column(ForeignKey('galaxy_group.id'), index=True)
create_time = Column(DateTime, default=now)
update_time = Column(DateTime, default=now, onupdate=now)
user = relationship('User', back_populates='groups')
group = relationship('Group', back_populates='users')
def __init__(self, user, group):
self.user = user
self.group = group
[docs]class UserRoleAssociation(Base, _HasTable):
__tablename__ = 'user_role_association'
id = Column(Integer, primary_key=True)
user_id = Column(ForeignKey('galaxy_user.id'), index=True)
role_id = Column(ForeignKey('role.id'), index=True)
create_time = Column(DateTime, default=now)
update_time = Column(DateTime, default=now, onupdate=now)
user = relationship('User', back_populates='roles')
role = relationship('Role', back_populates='users')
def __init__(self, user, role):
self.user = user
self.role = role
[docs]class GroupRoleAssociation(Base, _HasTable):
__tablename__ = 'group_role_association'
id = Column(Integer, primary_key=True)
group_id = Column(ForeignKey("galaxy_group.id"), index=True)
role_id = Column(ForeignKey("role.id"), index=True)
create_time = Column(DateTime, default=now)
update_time = Column(DateTime, default=now, onupdate=now)
group = relationship('Group', back_populates='roles')
role = relationship('Role', back_populates='groups')
def __init__(self, group, role):
self.group = group
self.role = role
[docs]class RepositoryRoleAssociation(Base, _HasTable):
__tablename__ = 'repository_role_association'
id = Column(Integer, primary_key=True)
repository_id = Column(ForeignKey("repository.id"), index=True)
role_id = Column(ForeignKey("role.id"), index=True)
create_time = Column(DateTime, default=now)
update_time = Column(DateTime, default=now, onupdate=now)
repository = relationship('Repository', back_populates='roles')
role = relationship('Role', back_populates='repositories')
def __init__(self, repository, role):
self.repository = repository
self.role = role
[docs]class GalaxySession(Base, _HasTable):
__tablename__ = 'galaxy_session'
id = Column(Integer, primary_key=True)
create_time = Column(DateTime, default=now)
update_time = Column(DateTime, default=now, onupdate=now)
user_id = Column(ForeignKey("galaxy_user.id"), index=True, nullable=True)
remote_host = Column(String(255))
remote_addr = Column(String(255))
referer = Column(TEXT)
# unique 128 bit random number coerced to a string
session_key = Column(TrimmedString(255), index=True, unique=True)
is_valid = Column(Boolean, default=False)
# saves a reference to the previous session so we have a way to chain them together
prev_session_id = Column(Integer)
last_action = Column(DateTime)
user = relationship('User', back_populates='galaxy_sessions')
def __init__(self, is_valid=False, **kwd):
super().__init__(**kwd)
self.is_valid = is_valid
self.last_action = self.last_action or datetime.now()
[docs]class Repository(Base, Dictifiable, _HasTable):
__tablename__ = 'repository'
id = Column(Integer, primary_key=True)
create_time = Column(DateTime, default=now)
update_time = Column(DateTime, default=now, onupdate=now)
name = Column(TrimmedString(255), index=True)
type = Column(TrimmedString(255), index=True)
remote_repository_url = Column(TrimmedString(255))
homepage_url = Column(TrimmedString(255))
description = Column(TEXT)
long_description = Column(TEXT)
user_id = Column(ForeignKey("galaxy_user.id"), index=True)
private = Column(Boolean, default=False)
deleted = Column(Boolean, index=True, default=False)
email_alerts = Column(MutableJSONType, nullable=True)
times_downloaded = Column(Integer)
deprecated = Column(Boolean, default=False)
categories = relationship('RepositoryCategoryAssociation', back_populates='repository')
ratings = relationship('RepositoryRatingAssociation',
order_by=lambda: desc(RepositoryRatingAssociation.update_time), back_populates='repository')
user = relationship('User', back_populates='active_repositories')
downloadable_revisions = relationship('RepositoryMetadata',
primaryjoin=lambda: (Repository.id == RepositoryMetadata.repository_id) & (RepositoryMetadata.downloadable == true()), # type: ignore[attr-defined,has-type]
viewonly=True,
order_by=lambda: desc(RepositoryMetadata.update_time)) # type: ignore[attr-defined]
metadata_revisions = relationship('RepositoryMetadata',
order_by=lambda: desc(RepositoryMetadata.update_time), # type: ignore[attr-defined]
back_populates='repository')
roles = relationship('RepositoryRoleAssociation', back_populates='repository')
reviews = relationship('RepositoryReview', back_populates='repository')
reviewers = relationship('User', secondary=lambda: RepositoryReview.__table__, viewonly=True) # type: ignore
dict_collection_visible_keys = ['id', 'name', 'type', 'remote_repository_url', 'homepage_url', 'description', 'user_id', 'private', 'deleted',
'times_downloaded', 'deprecated', 'create_time']
dict_element_visible_keys = ['id', 'name', 'type', 'remote_repository_url', 'homepage_url', 'description', 'long_description', 'user_id', 'private',
'deleted', 'times_downloaded', 'deprecated', 'create_time']
file_states = Bunch(NORMAL='n',
NEEDS_MERGING='m',
MARKED_FOR_REMOVAL='r',
MARKED_FOR_ADDITION='a',
NOT_TRACKED='?')
def __init__(self, private=False, times_downloaded=0, deprecated=False, **kwd):
super().__init__(**kwd)
self.private = private
self.times_downloaded = times_downloaded
self.deprecated = deprecated
self.name = self.name or "Unnamed repository"
@property
def hg_repo(self):
from mercurial import (
hg,
ui
)
if not WEAK_HG_REPO_CACHE.get(self):
WEAK_HG_REPO_CACHE[self] = hg.cachedlocalrepo(hg.repository(ui.ui(), self.repo_path().encode('utf-8')))
return WEAK_HG_REPO_CACHE[self].fetch()[0]
@property
def admin_role(self):
admin_role_name = f'{str(self.name)}_{str(self.user.username)}_admin'
for rra in self.roles:
role = rra.role
if str(role.name) == admin_role_name:
return role
raise Exception('Repository %s owned by %s is not associated with a required administrative role.' %
(str(self.name), str(self.user.username)))
[docs] def allow_push(self):
hgrc_file = hg_util.get_hgrc_path(self.repo_path())
with open(hgrc_file) as fh:
for line in fh.read().splitlines():
if line.startswith('allow_push = '):
return line[len('allow_push = '):]
return ''
[docs] def can_change_type(self):
# Allow changing the type only if the repository has no contents, has never been installed, or has
# never been changed from the default type.
if self.is_new():
return True
if self.times_downloaded == 0:
return True
if self.type == rt_util.UNRESTRICTED:
return True
return False
[docs] def can_change_type_to(self, app, new_type_label):
if self.type == new_type_label:
return False
if self.can_change_type():
new_type = app.repository_types_registry.get_class_by_label(new_type_label)
if new_type.is_valid_for_type(self):
return True
return False
[docs] def get_changesets_for_setting_metadata(self, app):
type_class = self.get_type_class(app)
return type_class.get_changesets_for_setting_metadata(app, self)
[docs] def get_repository_dependencies(self, app, changeset, toolshed_url):
# We aren't concerned with repositories of type tool_dependency_definition here if a
# repository_metadata record is not returned because repositories of this type will never
# have repository dependencies. However, if a readme file is uploaded, or some other change
# is made that does not create a new downloadable changeset revision but updates the existing
# one, we still want to be able to get repository dependencies.
repository_metadata = metadata_util.get_current_repository_metadata_for_changeset_revision(app,
self,
changeset)
if repository_metadata:
metadata = repository_metadata.metadata
if metadata:
rb = relation_builder.RelationBuilder(app, self, repository_metadata, toolshed_url)
repository_dependencies = rb.get_repository_dependencies_for_changeset_revision()
if repository_dependencies:
return repository_dependencies
return None
[docs] def get_type_class(self, app):
return app.repository_types_registry.get_class_by_label(self.type)
[docs] def get_tool_dependencies(self, app, changeset_revision):
changeset_revision = metadata_util.get_next_downloadable_changeset_revision(app, self, changeset_revision)
for downloadable_revision in self.downloadable_revisions:
if downloadable_revision.changeset_revision == changeset_revision:
return downloadable_revision.metadata.get('tool_dependencies', {})
return {}
[docs] def installable_revisions(self, app, sort_revisions=True):
return metadata_util.get_metadata_revisions(app, self, sort_revisions=sort_revisions)
[docs] def repo_path(self, app=None):
# Keep app argument for compatibility with tool_shed_install Repository model
return hgweb_config_manager.get_entry(os.path.join("repos", self.user.username, self.name))
[docs] def revision(self):
repo = self.hg_repo
tip_ctx = repo[repo.changelog.tip()]
return f"{str(tip_ctx.rev())}:{str(tip_ctx)}"
[docs] def set_allow_push(self, usernames, remove_auth=''):
allow_push = util.listify(self.allow_push())
if remove_auth:
allow_push.remove(remove_auth)
else:
for username in util.listify(usernames):
if username not in allow_push:
allow_push.append(username)
allow_push = f"{','.join(allow_push)}\n"
# Why doesn't the following work?
# repo.ui.setconfig('web', 'allow_push', allow_push)
repo_dir = self.repo_path()
hgrc_file = hg_util.get_hgrc_path(repo_dir)
with open(hgrc_file) as fh:
lines = fh.readlines()
with open(hgrc_file, 'w') as fh:
for line in lines:
if line.startswith('allow_push'):
fh.write(f'allow_push = {allow_push}')
else:
fh.write(line)
[docs] def to_dict(self, view='collection', value_mapper=None):
rval = super().to_dict(view=view, value_mapper=value_mapper)
if 'user_id' in rval:
rval['owner'] = self.user.username
return rval
[docs]class RepositoryReview(Base, Dictifiable, _HasTable):
__tablename__ = 'repository_review'
id = Column(Integer, primary_key=True)
create_time = Column(DateTime, default=now)
update_time = Column(DateTime, default=now, onupdate=now)
repository_id = Column(ForeignKey('repository.id'), index=True)
changeset_revision = Column(TrimmedString(255), index=True)
user_id = Column(ForeignKey('galaxy_user.id'), index=True, nullable=False)
approved = Column(TrimmedString(255))
rating = Column(Integer, index=True)
deleted = Column(Boolean, index=True, default=False)
repository = relationship('Repository', back_populates='reviews')
# Take care when using the mapper below! It should be used only when a new review is being created for a repository change set revision.
# Keep in mind that repository_metadata records can be removed from the database for certain change set revisions when metadata is being
# reset on a repository!
repository_metadata = relationship('RepositoryMetadata',
viewonly=True,
foreign_keys=lambda: [RepositoryReview.repository_id, RepositoryReview.changeset_revision],
primaryjoin=lambda: ((RepositoryReview.repository_id == RepositoryMetadata.repository_id) # type: ignore[has-type]
& (RepositoryReview.changeset_revision == RepositoryMetadata.changeset_revision)), # type: ignore[has-type]
back_populates='reviews')
user = relationship('User', back_populates='repository_reviews')
component_reviews = relationship('ComponentReview',
viewonly=True,
primaryjoin=lambda: ((RepositoryReview.id == ComponentReview.repository_review_id) # type: ignore[has-type]
& (ComponentReview.deleted == false())), # type: ignore[has-type]
back_populates='repository_review')
private_component_reviews = relationship('ComponentReview',
viewonly=True,
primaryjoin=lambda: ((RepositoryReview.id == ComponentReview.repository_review_id) # type: ignore[has-type]
& (ComponentReview.deleted == false()) & (ComponentReview.private == true()))) # type: ignore[has-type]
dict_collection_visible_keys = ['id', 'repository_id', 'changeset_revision', 'user_id', 'rating', 'deleted']
dict_element_visible_keys = ['id', 'repository_id', 'changeset_revision', 'user_id', 'rating', 'deleted']
approved_states = Bunch(NO='no', YES='yes')
def __init__(self, deleted=False, **kwd):
super().__init__(**kwd)
self.deleted = deleted
[docs]class ComponentReview(Base, Dictifiable, _HasTable):
__tablename__ = 'component_review'
id = Column(Integer, primary_key=True)
create_time = Column(DateTime, default=now)
update_time = Column(DateTime, default=now, onupdate=now)
repository_review_id = Column(ForeignKey("repository_review.id"), index=True)
component_id = Column(ForeignKey("component.id"), index=True)
comment = Column(TEXT)
private = Column(Boolean, default=False)
approved = Column(TrimmedString(255))
rating = Column(Integer)
deleted = Column(Boolean, index=True, default=False)
repository_review = relationship('RepositoryReview', back_populates='component_reviews')
component = relationship('Component')
dict_collection_visible_keys = ['id', 'repository_review_id', 'component_id', 'private', 'approved', 'rating', 'deleted']
dict_element_visible_keys = ['id', 'repository_review_id', 'component_id', 'private', 'approved', 'rating', 'deleted']
approved_states = Bunch(NO='no', YES='yes', NA='not_applicable')
def __init__(self, private=False, approved=False, deleted=False, **kwd):
super().__init__(**kwd)
self.private = private
self.approved = approved
self.deleted = deleted
[docs]class Component(Base, _HasTable):
__tablename__ = 'component'
id = Column(Integer, primary_key=True)
name = Column(TrimmedString(255))
description = Column(TEXT)
[docs]class ItemRatingAssociation(_HasTable):
[docs] def __init__(self, id=None, user=None, item=None, rating=0, comment=''):
self.id = id
self.user = user
self.item = item
self.rating = rating
self.comment = comment
[docs]class RepositoryRatingAssociation(Base, ItemRatingAssociation, _HasTable):
__tablename__ = 'repository_rating_association'
id = Column(Integer, primary_key=True)
create_time = Column(DateTime, default=now)
update_time = Column(DateTime, default=now, onupdate=now)
repository_id = Column(ForeignKey("repository.id"), index=True)
user_id = Column(ForeignKey("galaxy_user.id"), index=True)
rating = Column(Integer, index=True)
comment = Column(TEXT)
repository = relationship('Repository', back_populates='ratings')
user = relationship('User')
[docs]class Category(Base, Dictifiable, _HasTable):
__tablename__ = 'category'
id = Column(Integer, primary_key=True)
create_time = Column(DateTime, default=now)
update_time = Column(DateTime, default=now, onupdate=now)
name = Column(TrimmedString(255), index=True, unique=True)
description = Column(TEXT)
deleted = Column(Boolean, index=True, default=False)
repositories = relationship('RepositoryCategoryAssociation', back_populates='category')
dict_collection_visible_keys = ['id', 'name', 'description', 'deleted']
dict_element_visible_keys = ['id', 'name', 'description', 'deleted']
def __init__(self, deleted=False, **kwd):
super().__init__(**kwd)
self.deleted = deleted
[docs]class RepositoryCategoryAssociation(Base, _HasTable):
__tablename__ = 'repository_category_association'
id = Column(Integer, primary_key=True)
repository_id = Column(ForeignKey('repository.id'), index=True)
category_id = Column(ForeignKey('category.id'), index=True)
category = relationship('Category', back_populates='repositories')
repository = relationship('Repository', back_populates='categories')
def __init__(self, repository=None, category=None):
self.repository = repository
self.category = category
[docs]class Tag(Base, _HasTable):
__tablename__ = 'tag'
__table_args__ = (
UniqueConstraint('name'),
)
id = Column(Integer, primary_key=True)
type = Column(Integer)
parent_id = Column(ForeignKey('tag.id'))
name = Column(TrimmedString(255))
children = relationship('Tag', back_populates='parent')
parent = relationship('Tag', back_populates='children', remote_side=[id])
def __str__(self):
return "Tag(id=%s, type=%i, parent_id=%s, name=%s)" % (self.id, self.type, self.parent_id, self.name)
# The RepositoryMetadata model is mapped imperatively (for details see discussion in PR #12064).
# TLDR: a declaratively-mapped class cannot have a .metadata attribute (it is used by SQLAlchemy's DeclarativeBase).
[docs]class RepositoryMetadata(Dictifiable, _HasTable):
# Once the class has been mapped, all Column items in this table will be available
# as instrumented class attributes on RepositoryMetadata.
table = Table('repository_metadata', mapper_registry.metadata,
Column('id', Integer, primary_key=True),
Column('create_time', DateTime, default=now),
Column('update_time', DateTime, default=now, onupdate=now),
Column('repository_id', ForeignKey('repository.id'), index=True),
Column('changeset_revision', TrimmedString(255), index=True),
Column('numeric_revision', Integer, index=True),
Column('metadata', MutableJSONType, nullable=True),
Column('tool_versions', MutableJSONType, nullable=True),
Column('malicious', Boolean, default=False),
Column('downloadable', Boolean, default=True),
Column('missing_test_components', Boolean, default=False, index=True),
Column('has_repository_dependencies', Boolean, default=False, index=True),
Column('includes_datatypes', Boolean, default=False, index=True),
Column('includes_tools', Boolean, default=False, index=True),
Column('includes_tool_dependencies', Boolean, default=False, index=True),
Column('includes_workflows', Boolean, default=False, index=True))
dict_collection_visible_keys = ['id', 'repository_id', 'numeric_revision', 'changeset_revision', 'malicious', 'downloadable', 'missing_test_components',
'has_repository_dependencies', 'includes_datatypes', 'includes_tools', 'includes_tool_dependencies',
'includes_tools_for_display_in_tool_panel', 'includes_workflows']
dict_element_visible_keys = ['id', 'repository_id', 'numeric_revision', 'changeset_revision', 'malicious', 'downloadable', 'missing_test_components',
'has_repository_dependencies', 'includes_datatypes', 'includes_tools', 'includes_tool_dependencies',
'includes_tools_for_display_in_tool_panel', 'includes_workflows', 'repository_dependencies']
def __init__(self, id=None, repository_id=None, numeric_revision=None, changeset_revision=None, metadata=None, tool_versions=None, malicious=False,
downloadable=False, missing_test_components=None, tools_functionally_correct=False, test_install_error=False,
has_repository_dependencies=False, includes_datatypes=False, includes_tools=False, includes_tool_dependencies=False,
includes_workflows=False):
self.id = id
self.repository_id = repository_id
self.numeric_revision = numeric_revision
self.changeset_revision = changeset_revision
self.metadata = metadata
self.tool_versions = tool_versions
self.malicious = malicious
self.downloadable = downloadable
self.missing_test_components = missing_test_components
self.has_repository_dependencies = has_repository_dependencies
# We don't consider the special case has_repository_dependencies_only_if_compiling_contained_td here.
self.includes_datatypes = includes_datatypes
self.includes_tools = includes_tools
self.includes_tool_dependencies = includes_tool_dependencies
self.includes_workflows = includes_workflows
@property
def includes_tools_for_display_in_tool_panel(self):
if self.metadata:
tool_dicts = self.metadata.get('tools', [])
for tool_dict in tool_dicts:
if tool_dict.get('add_to_tool_panel', True):
return True
return False
@property
def repository_dependencies(self):
if self.has_repository_dependencies:
return [repository_dependency for repository_dependency in self.metadata['repository_dependencies']['repository_dependencies']]
return []
# After the map_imperatively statement has been executed, the members of the
# properties dictionary (repository, reviews) will be available as instrumented
# class attributes on RepositoryMetadata.
mapper_registry.map_imperatively(RepositoryMetadata, RepositoryMetadata.table, properties=dict(
repository=relationship(Repository, back_populates='metadata_revisions'),
reviews=relationship(RepositoryReview,
viewonly=True,
foreign_keys=lambda: [RepositoryReview.repository_id, RepositoryReview.changeset_revision],
primaryjoin=lambda: ((RepositoryReview.repository_id == RepositoryMetadata.repository_id)
& (RepositoryReview.changeset_revision == RepositoryMetadata.changeset_revision)),
back_populates='repository_metadata')))
# Utility methods
[docs]def sort_by_attr(seq, attr):
"""
Sort the sequence of objects by object's attribute
Arguments:
seq - the list or any sequence (including immutable one) of objects to sort.
attr - the name of attribute to sort by
"""
# Use the "Schwartzian transform"
# Create the auxiliary list of tuples where every i-th tuple has form
# (seq[i].attr, i, seq[i]) and sort it. The second item of tuple is needed not
# only to provide stable sorting, but mainly to eliminate comparison of objects
# (which can be expensive or prohibited) in case of equal attribute values.
intermed = [(getattr(v, attr), i, v) for i, v in enumerate(seq)]
intermed.sort()
return [_[-1] for _ in intermed]