Warning

This document is for an old release 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.model.orm.util import add_object_to_object_session
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)
[docs] def get_disk_usage(self, nice_size=False): return 0
@property def nice_total_disk_usage(self): return 0
[docs] def set_disk_usage(self, bytes): pass
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() add_object_to_object_session(self, user) 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): add_object_to_object_session(self, user) 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): add_object_to_object_session(self, user) self.user = user add_object_to_object_session(self, role) 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): add_object_to_object_session(self, repository) 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 is_new(self): tip_rev = self.hg_repo.changelog.tiprev() return tip_rev < 0
[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 tip(self): repo = self.hg_repo return str(repo[repo.changelog.tip()])
[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()) # type: ignore[has-type] & (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] def set_item(self, item): """Set association's item."""
[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] def set_item(self, repository): self.repository = repository
[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]