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 #15247 from jmchilton/tool_shed_cleanup_0

Browse Source 
Various Tool Shed Cleanup
This commit is contained in:
John Chilton
2023-02-07 16:23:18 -05:00
committed by GitHub
parent 8013e03423 e9832bfbfb
commit ae2e5c5341
28 changed files with 488 additions and 643 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/tool_shed/galaxy_install/install_manager.py
+9 -5
View File
@@ -36,6 +36,10 @@ from galaxy.util.tool_shed import (
common_util,
encoding_util,
)
from tool_shed_client.schema import (
ExtraRepoInfo,
RepositoryMetadataInstallInfoDict,
)
log = logging.getLogger(__name__)
@@ -73,7 +77,7 @@ class InstallRepositoryManager:
def __get_install_info_from_tool_shed(
self, tool_shed_url: str, name: str, owner: str, changeset_revision: str
) -> Tuple[Dict[str, Any], List[Dict[str, Any]]]:
) -> Tuple[RepositoryMetadataInstallInfoDict, List[ExtraRepoInfo]]:
params = dict(name=name, owner=owner, changeset_revision=changeset_revision)
pathspec = ["api", "repositories", "get_repository_revision_install_info"]
try:
@@ -94,8 +98,8 @@ class InstallRepositoryManager:
# Repository revision (RepositoryMetadata), and a dictionary including the additional
# information required to install the repository.
items = json.loads(util.unicodify(raw_text))
repository_revision_dict = items[1]
repo_info_dict = items[2]
repository_revision_dict: RepositoryMetadataInstallInfoDict = items[1]
repo_info_dict: ExtraRepoInfo = items[2]
else:
message = (
"Unable to retrieve installation information from tool shed %s for revision %s of repository %s owned by %s"
@@ -343,8 +347,8 @@ class InstallRepositoryManager:
def __initiate_and_install_repositories(
self,
tool_shed_url: str,
repository_revision_dict: Dict[str, Any],
repo_info_dicts: List[Dict[str, Any]],
repository_revision_dict: RepositoryMetadataInstallInfoDict,
repo_info_dicts: List[ExtraRepoInfo],
install_options: Dict[str, Any],
):
try:
lib/galaxy/tool_shed/util/tool_util.py
-34
View File
@@ -1,18 +1,13 @@
import logging
import os
import shutil
from typing import Optional
import galaxy.tools
from galaxy import util
from galaxy.datatypes.sniff import is_column_based
from galaxy.tool_shed.util import basic_util
from galaxy.util import checkers
from galaxy.util.expressions import ExpressionContext
from galaxy.web.form_builder import SelectField
log = logging.getLogger(__name__)
def build_shed_tool_conf_select_field(app):
"""Build a SelectField whose options are the keys in app.toolbox.shed_tool_confs."""
@@ -197,34 +192,6 @@ def _is_data_index_sample_file(file_path):
return True
def new_state(trans, tool, invalid=False):
"""Create a new `DefaultToolState` for the received tool. Only inputs on the first page will be initialized."""
state = galaxy.tools.DefaultToolState()
state.inputs = {}
if invalid:
# We're attempting to display a tool in the tool shed that has been determined to have errors, so is invalid.
return state
try:
# Attempt to generate the tool state using the standard Galaxy-side code
return tool.new_state(trans)
except Exception as e:
# Fall back to building tool state as below
log.debug(
'Failed to build tool state for tool "%s" using standard method, will try to fall back on custom method: %s',
tool.id,
e,
)
inputs = tool.inputs_by_page[0]
context = ExpressionContext(state.inputs, parent=None)
for input in inputs.values():
try:
state.inputs[input.name] = input.get_initial_value(trans, context)
except Exception:
# FIXME: not all values should be an empty list
state.inputs[input.name] = []
return state
def panel_entry_per_tool(tool_section_dict):
# Return True if tool_section_dict looks like this.
# {<Tool guid> :
@@ -251,6 +218,5 @@ __all__ = (
"copy_sample_files",
"generate_message_for_invalid_tools",
"handle_missing_index_file",
"new_state",
"panel_entry_per_tool",
)
lib/galaxy/util/tool_shed/tool_shed_registry.py
+9 -2
View File
@@ -68,12 +68,19 @@ class Registry:
Following more what galaxy.demo_sequencer.controllers.common does might be more appropriate at
some stage...
"""
shed_name = self._shed_name_for_url(url)
if shed_name is not None:
return self.tool_sheds_auth[shed_name]
else:
log.debug(f"Invalid url '{str(url)}' received by tool shed registry's url_auth method.")
return None
def _shed_name_for_url(self, url: str) -> Optional[str]:
url_sans_protocol = common_util.remove_protocol_from_tool_shed_url(url)
for shed_name, shed_url in self.tool_sheds.items():
shed_url_sans_protocol = common_util.remove_protocol_from_tool_shed_url(shed_url)
if url_sans_protocol.startswith(shed_url_sans_protocol):
return self.tool_sheds_auth[shed_name]
log.debug(f"Invalid url '{str(url)}' received by tool shed registry's url_auth method.")
return shed_name
return None
def get_tool_shed_url(self, tool_shed: str) -> Optional[str]:
lib/galaxy/webapps/galaxy/api/users.py
+1 -1
View File
@@ -76,7 +76,7 @@ APIKeyPathParam: str = Path(..., title="API Key", description="The API key of th
@router.cbv
class FastAPIHistories:
class FastAPIUsers:
service: UsersService = depends(UsersService)
@router.put(
lib/tool_shed/context.py
+86
View File
@@ -0,0 +1,86 @@
import abc
from typing import Optional
from sqlalchemy.orm import scoped_session
from galaxy.security.idencoding import IdEncodingHelper
from galaxy.work.context import (
GalaxyAbstractRequest,
GalaxyAbstractResponse,
)
from tool_shed.structured_app import ToolShedApp
from tool_shed.webapp.model import (
GalaxySession,
User,
)
from tool_shed.webapp.model.mapping import ToolShedModelMapping
class ProvidesAppContext:
"""For transaction-like objects to provide the shed convenience layer for
database and event handling.
Mixed in class must provide `app` property.
"""
@abc.abstractproperty
def app(self) -> ToolShedApp:
"""Provide access to the shed ``app`` object."""
@property
def security(self) -> IdEncodingHelper:
return self.app.security
@property
def sa_session(self) -> scoped_session:
"""Provide access to Galaxy's SQLAlchemy session.
:rtype: galaxy.model.scoped_session.galaxy_scoped_session
"""
return self.model.session
@property
def model(self) -> ToolShedModelMapping:
"""Provide access to Tool Shed's model mapping class."""
return self.app.model
class ProvidesUserContext(ProvidesAppContext):
"""For transaction-like objects to provide Galaxy convenience layer for
reasoning about users.
Mixed in class must provide `user` and `app`
properties.
"""
@abc.abstractproperty
def user(self) -> Optional[User]:
"""Provide access to the user object."""
@property
def anonymous(self) -> bool:
return self.user is None
@property
def user_is_admin(self) -> bool:
return self.app.config.is_admin_user(self.user)
@property
def user_is_bootstrap_admin(self) -> bool:
"""Master key provided so there is no real user"""
user = self.user
return not self.anonymous and user is not None and user.bootstrap_admin_user
class SessionRequestContext(ProvidesUserContext):
@abc.abstractmethod
def get_galaxy_session(self) -> Optional[GalaxySession]:
...
@abc.abstractproperty
def request(self) -> GalaxyAbstractRequest:
...
@abc.abstractproperty
def response(self) -> GalaxyAbstractResponse:
...
lib/tool_shed/dependencies/repository/relation_builder.py
+5 -14
View File
@@ -25,11 +25,6 @@ class RelationBuilder:
self.handled_key_rd_dicts = []
self.key_rd_dicts_to_be_processed = []
self.tool_shed_url = tool_shed_url
# This is a temporary work-around for handling repository dependencies that are needed
# only if compiling a dependent package. This value should be True unless exporting
# a repository capsule, in which case the set_filter_dependencies_needed_for_compiling()
# function is called.
self.filter_dependencies_needed_for_compiling = True
def can_add_to_key_rd_dicts(self, key_rd_dict, key_rd_dicts):
"""Handle the case where an update to the changeset revision was done."""
@@ -440,12 +435,11 @@ class RelationBuilder:
current_repository_key_rd_dicts
)
for key_rd_dict in current_repository_key_rd_dicts:
if self.filter_dependencies_needed_for_compiling:
# Filter out repository dependencies that are required only if compiling the dependent
# repository's tool dependency.
# TODO: this temporary work-around should be removed when the underlying framework
# support for handling only_if_compiling_contained_td-flagged repositories is completed.
key_rd_dict = self.filter_only_if_compiling_contained_td(key_rd_dict)
# Filter out repository dependencies that are required only if compiling the dependent
# repository's tool dependency.
# TODO: this temporary work-around should be removed when the underlying framework
# support for handling only_if_compiling_contained_td-flagged repositories is completed.
key_rd_dict = self.filter_only_if_compiling_contained_td(key_rd_dict)
if key_rd_dict:
is_circular = False
in_handled_key_rd_dicts = self.in_key_rd_dicts(key_rd_dict, self.handled_key_rd_dicts)
@@ -544,9 +538,6 @@ class RelationBuilder:
clean_key_rd_dicts.append(new_key_rd_dict)
return clean_key_rd_dicts
def set_filter_dependencies_needed_for_compiling(self, value):
self.filter_dependencies_needed_for_compiling = asbool(value)
def update_circular_repository_dependencies(self, repository_key, repository_dependency, repository_dependencies):
repository_key_as_repository_dependency = repository_key.split(container_util.STRSEP)
if repository_key_as_repository_dependency in repository_dependencies:
lib/tool_shed/structured_app.py
+4
View File
@@ -3,10 +3,14 @@ from typing import TYPE_CHECKING
from galaxy.structured_app import BasicSharedApp
if TYPE_CHECKING:
from tool_shed.repository_registry import Registry as RepositoryRegistry
from tool_shed.repository_types.registry import Registry as RepositoryTypesRegistry
from tool_shed.util.hgweb_config import HgWebConfigManager
from tool_shed.webapp.model import mapping
class ToolShedApp(BasicSharedApp):
repository_types_registry: "RepositoryTypesRegistry"
model: "mapping.ToolShedModelMapping"
repository_registry: "RepositoryRegistry"
hgweb_config_manager: "HgWebConfigManager"
lib/tool_shed/test/base/populators.py
+34 -19
View File
@@ -15,7 +15,11 @@ from galaxy_test.base.api_util import random_name
from tool_shed_client.schema import (
Category,
CreateCategoryRequest,
CreateRepositoryRequest,
from_legacy_install_info,
GetInstallInfoRequest,
GetOrderedInstallableRevisionsRequest,
InstallInfo,
OrderedInstallableRevisions,
Repository,
RepositoryIndexRequest,
@@ -24,6 +28,7 @@ from tool_shed_client.schema import (
RepositorySearchRequest,
RepositorySearchResults,
RepositoryUpdate,
RepositoryUpdateRequest,
ResetMetadataOnRepositoryRequest,
ResetMetadataOnRepositoryResponse,
ToolSearchRequest,
@@ -66,6 +71,20 @@ class ToolShedPopulator:
repository = self.setup_column_maker_repo(prefix=prefix)
return self.get_metadata(repository)
def get_install_info(self, repository_metadata: RepositoryMetadata) -> InstallInfo:
revision_metadata = repository_metadata.latest_revision
repo = revision_metadata.repository
request = GetInstallInfoRequest(
owner=repo.owner,
name=repo.name,
changeset_revision=revision_metadata.changeset_revision,
)
revisions_response = self._api_interactor.get(
"repositories/get_repository_revision_install_info", params=request.dict()
)
api_asserts.assert_status_code_is_ok(revisions_response)
return from_legacy_install_info(revisions_response.json())
def update_column_maker_repo(self, repository: HasRepositoryId) -> requests.Response:
response = self.upload_revision(
repository,
@@ -76,12 +95,14 @@ class ToolShedPopulator:
def upload_revision_raw(
self, repository: HasRepositoryId, path: Traversable, commit_message: str = DEFAULT_COMMIT_MESSAGE
) -> requests.Response:
body = {
"commit_message": commit_message,
}
body = RepositoryUpdateRequest(
commit_message=commit_message,
)
files = {"file": path.open("rb")}
repository_id = self._repository_id(repository)
response = self._api_interactor.post(f"repositories/{repository_id}/changeset_revision", json=body, files=files)
response = self._api_interactor.post(
f"repositories/{repository_id}/changeset_revision", params=body.dict(), files=files
)
return response
def upload_revision(
@@ -94,21 +115,15 @@ class ToolShedPopulator:
def new_repository(self, category_id, prefix=DEFAULT_PREFIX) -> Repository:
name = random_name(prefix=prefix)
synopsis = random_name(prefix=prefix)
description = None
remote_repository_url = None
homepage_url = None
category_ids = category_id
type = "unrestricted"
body = {
"name": name,
"synopsis": synopsis,
"description": description,
"remote_repository_url": remote_repository_url,
"homepage_url": homepage_url,
"category_ids[]": category_ids,
"type": type,
}
response = self._api_interactor.post("repositories", json=body)
request = CreateRepositoryRequest(
name=name,
synopsis=synopsis,
category_ids=category_id,
)
return self.create_repository(request)
def create_repository(self, request: CreateRepositoryRequest) -> Repository:
response = self._api_interactor.post("repositories", json=request.dict(by_alias=True))
api_asserts.assert_status_code_is_ok(response)
return Repository(**response.json())
lib/tool_shed/test/base/twilltestcase.py
-29
View File
@@ -297,20 +297,6 @@ class ShedTwillTestCase(ShedApiTestCase):
self.check_repository_changelog(repository)
self.check_string_count_in_page("Repository metadata is associated with this change set.", metadata_count)
def check_exported_repository_dependency(self, dependency_filename, repository_name, repository_owner):
root, error_message = xml_util.parse_xml(dependency_filename)
for elem in root.findall("repository"):
if "changeset_revision" in elem:
raise AssertionError(
"Exported repository %s with owner %s has a dependency with a defined changeset revision."
% (repository_name, repository_owner)
)
if "toolshed" in elem:
raise AssertionError(
"Exported repository %s with owner %s has a dependency with a defined tool shed."
% (repository_name, repository_owner)
)
def check_for_valid_tools(self, repository, strings_displayed=None, strings_not_displayed=None):
if strings_displayed is None:
strings_displayed = ["Valid tools"]
@@ -810,13 +796,6 @@ class ShedTwillTestCase(ShedApiTestCase):
os.makedirs(temp_path)
return temp_path
def get_datatypes_count(self):
params = {"upload_only": False}
self.visit_galaxy_url("/api/datatypes", params=params)
html = self.last_page()
datatypes = loads(html)
return len(datatypes)
def get_filename(self, filename, filepath=None):
if filepath is not None:
return os.path.abspath(os.path.join(filepath, filename))
@@ -1357,14 +1336,6 @@ class ShedTwillTestCase(ShedApiTestCase):
assert "The status has not changed in the tool shed for repository" in message, str(response_dict)
return response_dict
def update_tool_shed_status(self):
api_key = get_admin_api_key()
response = requests.get(
f"{self.galaxy_url}/api/tool_shed_repositories/check_for_updates?key={api_key}",
timeout=DEFAULT_SOCKET_TIMEOUT,
)
assert response.status_code != 403, response.content
def upload_file(
self,
repository: Repository,
lib/tool_shed/test/functional/test_0540_get_all_metadata_from_api.py
-170
View File
@@ -1,170 +0,0 @@
import logging
from ..base.twilltestcase import (
common,
ShedTwillTestCase,
)
log = logging.getLogger(__name__)
repositories = dict(
column=dict(
name="column_maker_0540",
description="Description for column_maker_0540",
long_description="Long description for column_maker_0540",
),
convert=dict(
name="convert_chars_0540",
description="Description for convert_chars_0540",
long_description="Long description for convert_chars_0540",
),
bwa=dict(
name="package_bwa_0_5_9_0540",
description="Description for package_bwa_0_5_9_0540",
long_description="Long description for package_bwa_0_5_9_0540",
),
)
category_name = "Test 0540"
category_description = "Verify API endpoint to retrieve all metadata"
"""
1. Create repository column_maker_0540 as user user1.
2. Create repository convert_chars_0540 with dependency on column_maker_0540.
3. Create repository package_bwa_0_5_9_0540.
4. Create dependency on package_bwa_0_5_9_0540 for convert_chars_0540.
5. Load /api/repositories/convert_chars_0540.id/metadata and verify contents.
"""
class TestGetAllMetadata(ShedTwillTestCase):
"""Verify that the code correctly handles the repository admin role."""
def test_0000_initiate_users(self):
"""Create necessary user accounts."""
self.login(email=common.test_user_1_email, username=common.test_user_1_name)
self.login(email=common.admin_email, username=common.admin_username)
def test_0005_create_bwa_package_repository(self):
"""Create and populate package_bwa_0_5_9_0540."""
category = self.create_category(name=category_name, description=category_description)
self.login(email=common.test_user_1_email, username=common.test_user_1_name)
# Create a repository named package_bwa_0_5_9_0100 owned by user1.
repository = self.get_or_create_repository(
name=repositories["bwa"]["name"],
description=repositories["bwa"]["description"],
long_description=repositories["bwa"]["long_description"],
owner=common.test_user_1_name,
category=category,
strings_displayed=[],
)
assert repository is not None, f"Error creating repository {repositories['bwa']['name']}"
self.upload_file(
repository,
filename="0540_files/package_bwa/tool_dependencies.xml",
filepath=None,
valid_tools_only=True,
uncompress_file=False,
remove_repo_files_not_in_tar=False,
commit_message="Uploaded tool_dependencies.xml.",
strings_displayed=["This repository currently contains a single file named <b>tool_dependencies.xml</b>"],
strings_not_displayed=[],
)
# Visit the manage repository page for package_bwa_0_5_9_0100.
self.display_manage_repository_page(
repository, strings_displayed=["Tool dependencies", "will not be", "to this repository"]
)
def test_0010_create_convert_repository(self):
"""Create the convert_chars_0540 repository."""
category = self.create_category(name=category_name, description=category_description)
repository = self.get_or_create_repository(
name=repositories["convert"]["name"],
description=repositories["convert"]["description"],
long_description=repositories["convert"]["long_description"],
owner=common.test_user_1_name,
category=category,
strings_displayed=[],
)
assert repository is not None, f"Error creating repository {repositories['convert']['name']}"
self.upload_file(
repository,
filename="0540_files/convert_chars/convert_chars.tar",
filepath=None,
valid_tools_only=True,
uncompress_file=True,
remove_repo_files_not_in_tar=False,
commit_message="Uploaded column maker 1.0.",
strings_displayed=[],
strings_not_displayed=[],
)
# Add a dependency on BWA.
self.upload_file(
repository,
filename="0540_files/convert_chars/tool_dependencies.xml",
filepath=None,
valid_tools_only=True,
uncompress_file=True,
remove_repo_files_not_in_tar=False,
commit_message="Uploaded column maker 1.0.",
strings_displayed=[],
strings_not_displayed=[],
)
# Visit the manage repository page for convert_chars_0540.
self.display_manage_repository_page(repository, strings_displayed=[repositories["bwa"]["name"]])
def test_0015_create_column_repository(self):
"""Create the column_maker_0540 repository."""
category = self.create_category(name=category_name, description=category_description)
repository = self.get_or_create_repository(
name=repositories["column"]["name"],
description=repositories["column"]["description"],
long_description=repositories["column"]["long_description"],
owner=common.test_user_1_name,
category=category,
strings_displayed=[],
)
assert repository is not None, f"Error creating repository {repositories['column']['name']}"
self.upload_file(
repository,
filename="0540_files/column_maker/column_maker.tar",
filepath=None,
valid_tools_only=True,
uncompress_file=True,
remove_repo_files_not_in_tar=False,
commit_message="Uploaded column maker 1.0.",
strings_displayed=[],
strings_not_displayed=[],
)
def test_0020_create_repository_dependency(self):
"""Make column_maker depend on convert_chars."""
repository = self._get_repository_by_name_and_owner(repositories["column"]["name"], common.test_user_1_name)
self.upload_file(
repository,
filename="0540_files/column_maker/repository_dependencies.xml",
filepath=None,
valid_tools_only=True,
uncompress_file=True,
remove_repo_files_not_in_tar=False,
commit_message="Uploaded column maker 2.0.",
strings_displayed=[],
strings_not_displayed=[],
)
def test_0025_verify_dependency_json(self):
"""
Load the API endpoint to retrieve all repository metadata and verify
that all three repository names are displayed.
"""
repository = self._get_repository_by_name_and_owner(repositories["column"]["name"], common.test_user_1_name)
strings_displayed = [
repositories["column"]["name"],
repositories["convert"]["name"],
repositories["bwa"]["name"],
]
self.fetch_repository_metadata(repository, strings_displayed=strings_displayed, strings_not_displayed=None)
lib/tool_shed/test/functional/test_shed_repositories.py
+8
View File
@@ -62,6 +62,14 @@ class TestShedRepositoriesApi(ShedApiTestCase):
assert repository.owner == repo.owner
assert repository.name == repo.name
def test_install_info(self):
# actually installing requires a whole Galaxy setup and the install manager but
# we can test the response validates against the future facing InstallInfo pydandic
# models.
populator = self.populator
repo = populator.setup_column_maker_and_get_metadata(prefix="repoforinstallinfo")
populator.get_install_info(repo)
def test_get_ordered_installable_revisions(self):
# Used in ephemeris...
populator = self.populator
lib/tool_shed/test/functional/test_shed_users.py
+11 -2
View File
@@ -3,6 +3,10 @@ from urllib.parse import urljoin
from requests import post
from galaxy_test.base import api_asserts
from tool_shed_client.schema import (
CreateUserRequest,
User,
)
from ..base.api import (
email_to_username,
ensure_user_with_email,
@@ -24,14 +28,19 @@ class TestShedUsersApi(ShedApiTestCase):
}
email = "testcreateuser@bx.psu.edu"
password = "mycoolpassword123"
username = email_to_username(email)
body = {
"email": email,
"password": password,
"username": email_to_username(email),
"username": username,
}
response = post(url, json=body, headers=headers)
request = CreateUserRequest(**body)
response = post(url, json=request.dict(), headers=headers)
api_asserts.assert_status_code_is_ok(response)
self._verify_username_password(email, password)
user = User(**response.json())
assert user.id
assert user.username == username
def test_create_user_interactor(self):
email = "testcreateuserinteractor@bx.psu.edu"
lib/tool_shed/test/test_data/0540_files/column_maker/column_maker.tar
BIN
View File
Binary file not shown.
lib/tool_shed/test/test_data/0540_files/column_maker/repository_dependencies.xml
-4
View File
@@ -1,4 +0,0 @@
<?xml version="1.0"?>
<repositories description="Simple dependency on convert_chars.">
<repository name="convert_chars_0540" owner="user1" />
</repositories>
lib/tool_shed/test/test_data/0540_files/convert_chars/convert_chars.tar
BIN
View File
Binary file not shown.
lib/tool_shed/test/test_data/0540_files/convert_chars/tool_dependencies.xml
-6
View File
@@ -1,6 +0,0 @@
<?xml version="1.0"?>
<tool_dependency>
<package name="bwa" version="0.5.9">
<repository name="package_bwa_0_5_9_0540" owner="user1" />
</package>
</tool_dependency>
lib/tool_shed/test/test_data/0540_files/package_bwa/tool_dependencies.xml
-13
View File
@@ -1,13 +0,0 @@
<?xml version="1.0"?>
<tool_dependency>
<package name="bwa" version="0.5.9">
<install version="1.0">
<actions>
<action type="download_by_url">file://__PATH__/bwa_base.tar</action>
<action type="set_environment">
<environment_variable name="PATH" action="prepend_to">$INSTALL_DIR/bin</environment_variable>
</action>
</actions>
</install>
</package>
</tool_dependency>
lib/tool_shed/tools/tool_version_manager.py
-27
View File
@@ -1,7 +1,5 @@
import logging
from sqlalchemy import and_
from tool_shed.util import (
hg_util,
metadata_util,
@@ -15,31 +13,6 @@ class ToolVersionManager:
def __init__(self, app):
self.app = app
def get_tool_version(self, tool_id):
context = self.app.install_model.context
return (
context.query(self.app.install_model.ToolVersion)
.filter(self.app.install_model.ToolVersion.table.c.tool_id == tool_id)
.first()
)
def get_tool_version_association(self, parent_tool_version, tool_version):
"""
Return a ToolVersionAssociation if one exists that associates the two
received tool_versions. This function is called only from Galaxy.
"""
context = self.app.install_model.context
return (
context.query(self.app.install_model.ToolVersionAssociation)
.filter(
and_(
self.app.install_model.ToolVersionAssociation.table.c.parent_id == parent_tool_version.id,
self.app.install_model.ToolVersionAssociation.table.c.tool_id == tool_version.id,
)
)
.first()
)
def get_version_lineage_for_tool(self, repository_id, repository_metadata, guid):
"""
Return the tool version lineage chain in descendant order for the received
lib/tool_shed/util/commit_util.py
+26 -12
View File
@@ -6,6 +6,13 @@ import os
import shutil
import tempfile
from collections import namedtuple
from typing import (
List,
Optional,
Tuple,
TYPE_CHECKING,
Union,
)
from sqlalchemy.sql.expression import null
@@ -19,13 +26,17 @@ from tool_shed.util import (
shed_util_common as suc,
)
if TYPE_CHECKING:
from tool_shed.structured_app import ToolShedApp
from tool_shed.webapp.model import Repository
log = logging.getLogger(__name__)
UNDESIRABLE_DIRS = [".hg", ".svn", ".git", ".cvs", ".idea"]
UNDESIRABLE_FILES = [".hg_archival.txt", "hgrc", ".DS_Store", "tool_test_output.html", "tool_test_output.json"]
def check_archive(repository, archive):
def check_archive(repository: "Repository", archive):
valid = []
invalid = []
errors = []
@@ -81,7 +92,7 @@ def check_archive(repository, archive):
return ArchiveCheckResults(valid, invalid, undesirable_files, undesirable_dirs, errors)
def check_file_contents_for_email_alerts(app):
def check_file_contents_for_email_alerts(app: "ToolShedApp"):
"""
See if any admin users have chosen to receive email alerts when a repository is updated.
If so, the file contents of the update must be checked for inappropriate content.
@@ -126,7 +137,7 @@ def get_change_lines_in_file_for_tag(tag, change_dict):
return cleaned_lines
def get_upload_point(repository, **kwd):
def get_upload_point(repository: "Repository", **kwd) -> Optional[str]:
upload_point = kwd.get("upload_point", None)
if upload_point is not None:
# The value of upload_point will be something like: database/community_files/000/repo_12/1.bed
@@ -148,7 +159,7 @@ def get_upload_point(repository, **kwd):
return upload_point
def handle_bz2(repository, uploaded_file_name):
def handle_bz2(repository: "Repository", uploaded_file_name):
with tempfile.NamedTemporaryFile(
mode="wb",
prefix=f"repo_{repository.id}_upload_bunzip2_",
@@ -168,19 +179,22 @@ def handle_bz2(repository, uploaded_file_name):
shutil.move(uncompressed.name, uploaded_file_name)
ChangeResponseT = Tuple[Union[bool, str], str, List[str], str, int, int]
def handle_directory_changes(
app,
host,
username,
repository,
full_path,
host: str,
username: str,
repository: "Repository",
full_path: str,
filenames_in_archive,
remove_repo_files_not_in_tar,
new_repo_alert,
commit_message,
undesirable_dirs_removed,
undesirable_files_removed,
):
commit_message: str,
undesirable_dirs_removed: int,
undesirable_files_removed: int,
) -> ChangeResponseT:
repo_path = repository.repo_path(app)
content_alert_str = ""
files_to_remove = []
lib/tool_shed/util/hg_util.py
+1 -1
View File
@@ -51,7 +51,7 @@ def archive_repository_revision(app, repository, archive_dir, changeset_revision
raise Exception(error_message)
def commit_changeset(repo_path, full_path_to_changeset, username, message):
def commit_changeset(repo_path: str, full_path_to_changeset: str, username: str, message: str) -> None:
try:
subprocess.check_output(
["hg", "commit", "-u", username, "-m", message, full_path_to_changeset],
lib/tool_shed/util/repository_util.py
+23 -11
View File
@@ -2,6 +2,11 @@ import configparser
import logging
import os
import re
from typing import (
Optional,
Tuple,
TYPE_CHECKING,
)
from markupsafe import escape
from sqlalchemy import false
@@ -54,13 +59,19 @@ from tool_shed.util.metadata_util import (
get_repository_metadata_by_changeset_revision,
)
if TYPE_CHECKING:
from tool_shed.context import ProvidesUserContext
from tool_shed.structured_app import ToolShedApp
from tool_shed.webapp.model import Repository
log = logging.getLogger(__name__)
VALID_REPOSITORYNAME_RE = re.compile(r"^[a-z0-9\_]+$")
def create_repo_info_dict(
app,
app: "ToolShedApp",
repository_clone_url,
changeset_revision,
ctx_rev,
@@ -139,7 +150,7 @@ def create_repo_info_dict(
return repo_info_dict
def create_repository_admin_role(app, repository):
def create_repository_admin_role(app: "ToolShedApp", repository: "Repository"):
"""
Create a new role with name-spaced name based on the repository name and its owner's public user
name. This will ensure that the tole name is unique.
@@ -160,16 +171,16 @@ def create_repository_admin_role(app, repository):
def create_repository(
app,
name,
type,
app: "ToolShedApp",
name: str,
type: str,
description,
long_description,
user_id,
category_ids=None,
remote_repository_url=None,
homepage_url=None,
):
) -> Tuple["Repository", str]:
"""Create a new ToolShed repository"""
category_ids = category_ids or []
sa_session = app.model.session
@@ -238,7 +249,7 @@ def get_repository_in_tool_shed(app, id, eagerload_columns=None):
return q.get(app.security.decode_id(id))
def get_repo_info_dict(app, user, repository_id, changeset_revision):
def get_repo_info_dict(app: "ToolShedApp", user, repository_id, changeset_revision):
repository = get_repository_in_tool_shed(app, repository_id)
repository_clone_url = common_util.generate_clone_url_for_repository_in_tool_shed(user, repository)
repository_metadata = get_repository_metadata_by_changeset_revision(app, repository_id, changeset_revision)
@@ -305,7 +316,7 @@ def get_repo_info_dict(app, user, repository_id, changeset_revision):
def get_repositories_by_category(
app, category_id, installable=False, sort_order="asc", sort_key="name", page=None, per_page=25
app: "ToolShedApp", category_id, installable=False, sort_order="asc", sort_key="name", page=None, per_page=25
):
sa_session = app.model.session
query = (
@@ -361,7 +372,7 @@ def get_repositories_by_category(
return repositories
def handle_role_associations(app, role, repository, **kwd):
def handle_role_associations(app: "ToolShedApp", role, repository, **kwd):
sa_session = app.model.session
message = escape(kwd.get("message", ""))
status = kwd.get("status", "done")
@@ -433,8 +444,9 @@ def change_repository_name_in_hgrc_file(hgrc_file, new_name):
config.write(fh)
def update_repository(app, trans, id, **kwds):
def update_repository(trans: "ProvidesUserContext", id: str, **kwds) -> Tuple[Optional["Repository"], Optional[str]]:
"""Update an existing ToolShed repository"""
app = trans.app
message = None
flush_needed = False
sa_session = app.model.session
@@ -507,7 +519,7 @@ def update_repository(app, trans, id, **kwds):
return repository, message
def validate_repository_name(app, name, user):
def validate_repository_name(app: "ToolShedApp", name, user):
"""
Validate whether the given name qualifies as a new TS repo name.
Repository names must be unique for each user, must be at least two characters
lib/tool_shed/util/shed_util_common.py
+13 -65
View File
@@ -3,6 +3,7 @@ import logging
import os
import socket
import string
from typing import TYPE_CHECKING
import sqlalchemy.orm.exc
from sqlalchemy import (
@@ -11,7 +12,6 @@ from sqlalchemy import (
true,
)
import galaxy.tool_util.deps.requirements
from galaxy import util
from galaxy.tool_shed.util.shed_util_common import (
can_eliminate_repository_dependency,
@@ -36,6 +36,9 @@ from tool_shed.util import (
repository_util,
)
if TYPE_CHECKING:
from tool_shed.structured_app import ToolShedApp
log = logging.getLogger(__name__)
MAX_CONTENT_SIZE = 1048576
@@ -99,7 +102,7 @@ This message was sent from the Galaxy Tool Shed instance hosted on the server
"""
def count_repositories_in_category(app, category_id):
def count_repositories_in_category(app: "ToolShedApp", category_id: str) -> int:
sa_session = app.model.session
return (
sa_session.query(app.model.RepositoryCategoryAssociation)
@@ -108,7 +111,7 @@ def count_repositories_in_category(app, category_id):
)
def get_categories(app):
def get_categories(app: "ToolShedApp"):
"""Get all categories from the database."""
sa_session = app.model.session
return (
@@ -119,13 +122,13 @@ def get_categories(app):
)
def get_category(app, id):
def get_category(app: "ToolShedApp", id: str):
"""Get a category from the database."""
sa_session = app.model.session
return sa_session.query(app.model.Category).get(app.security.decode_id(id))
def get_category_by_name(app, name):
def get_category_by_name(app: "ToolShedApp", name: str):
"""Get a category from the database via name."""
sa_session = app.model.session
try:
@@ -134,57 +137,6 @@ def get_category_by_name(app, name):
return None
def get_tool_shed_repo_requirements(app, tool_shed_url, repositories=None, repo_info_dicts=None):
"""
Contact tool_shed_url for a list of requirements for a repository or a list of repositories.
Returns a list of requirements, where each requirement is a dictionary with name and version as keys.
"""
if not repositories and not repo_info_dicts:
raise Exception("Need to pass either repository or repo_info_dicts")
if repositories:
if not isinstance(repositories, list):
repositories = [repositories]
repository_params = [
{"name": repository.name, "owner": repository.owner, "changeset_revision": repository.changeset_revision}
for repository in repositories
]
else:
if not isinstance(repo_info_dicts, list):
repo_info_dicts = [repo_info_dicts]
repository_params = []
for repo_info_dict in repo_info_dicts:
for name, repo_info_tuple in repo_info_dict.items():
# repo_info_tuple is a list, but keep terminology
owner = repo_info_tuple[4]
changeset_revision = repo_info_tuple[2]
repository_params.append({"name": name, "owner": owner, "changeset_revision": changeset_revision})
pathspec = ["api", "repositories", "get_repository_revision_install_info"]
tools = []
for params in repository_params:
response = util.url_get(
tool_shed_url, auth=app.tool_shed_registry.url_auth(tool_shed_url), pathspec=pathspec, params=params
)
json_response = json.loads(response)
valid_tools = json_response[1].get("valid_tools", [])
if valid_tools:
tools.extend(valid_tools)
return get_requirements_from_tools(tools)
def get_requirements_from_tools(tools):
return {
tool["id"]: galaxy.tool_util.deps.requirements.ToolRequirements.from_list(tool["requirements"])
for tool in tools
}
def get_requirements_from_repository(repository):
if not repository.includes_tools:
return {}
else:
return get_requirements_from_tools(repository.metadata_.get("tools", []))
def get_repository_categories(app, id):
"""Get categories of a repository on the tool shed side from the database via id"""
sa_session = app.model.session
@@ -196,10 +148,10 @@ def get_repository_categories(app, id):
def get_repository_file_contents(app, file_path, repository_id, is_admin=False):
"""Return the display-safe contents of a repository file for display in a browser."""
safe_str = ""
if not is_path_browsable(app, file_path, repository_id, is_admin):
if not _is_path_browsable(app, file_path, repository_id, is_admin):
log.warning("Request tries to access a file outside of the repository location. File path: %s", file_path)
return "Invalid file path"
# Symlink targets are checked by is_path_browsable
# Symlink targets are checked by _is_path_browsable
if os.path.islink(file_path):
safe_str = f"link to: {basic_util.to_html_string(os.readlink(file_path))}"
return safe_str
@@ -421,7 +373,7 @@ def handle_email_alerts(app, host, repository, content_alert_str="", new_repo_al
log.exception("An error occurred sending a tool shed repository update alert by email.")
def is_path_browsable(app, path, repository_id, is_admin=False):
def _is_path_browsable(app, path, repository_id, is_admin=False):
"""
Detects whether the given path is browsable i.e. is within the
allowed repository folders. Admins can additionaly browse folders
@@ -461,7 +413,7 @@ def open_repository_files_folder(app, folder_path, repository_id, is_admin=False
Return a list of dictionaries, each of which contains information for a file or directory contained
within a directory in a repository file hierarchy.
"""
if not is_path_browsable(app, folder_path, repository_id, is_admin):
if not _is_path_browsable(app, folder_path, repository_id, is_admin):
log.warning("Request tries to access a folder outside of the allowed locations. Folder path: %s", folder_path)
return []
try:
@@ -475,7 +427,7 @@ def open_repository_files_folder(app, folder_path, repository_id, is_admin=False
is_folder = False
full_path = os.path.join(folder_path, filename)
is_link = os.path.islink(full_path)
path_is_browsable = is_path_browsable(app, full_path, repository_id)
path_is_browsable = _is_path_browsable(app, full_path, repository_id)
if is_link and not path_is_browsable:
log.warning(
f"Valid folder contains a symlink outside of the repository location. Link found in: {str(full_path)}"
@@ -505,9 +457,6 @@ __all__ = (
"get_categories",
"get_category",
"get_category_by_name",
"get_requirements_from_tools",
"get_requirements_from_repository",
"get_tool_shed_repo_requirements",
"get_ctx_rev",
"get_next_prior_import_or_install_required_dict_entry",
"get_repository_categories",
@@ -519,7 +468,6 @@ __all__ = (
"get_user",
"handle_email_alerts",
"have_shed_tool_conf_for_install",
"is_path_browsable",
"is_path_within_dependency_dir",
"is_path_within_repo",
"open_repository_files_folder",
lib/tool_shed/util/tool_util.py
+35 -1
View File
@@ -1,9 +1,43 @@
import logging
import galaxy.tools
from galaxy.tool_shed.util.tool_util import (
copy_sample_file,
copy_sample_files,
generate_message_for_invalid_tools,
new_state,
)
from galaxy.util.expressions import ExpressionContext
log = logging.getLogger(__name__)
def new_state(trans, tool, invalid=False):
"""Create a new `DefaultToolState` for the received tool. Only inputs on the first page will be initialized."""
state = galaxy.tools.DefaultToolState()
state.inputs = {}
if invalid:
# We're attempting to display a tool in the tool shed that has been determined to have errors, so is invalid.
return state
try:
# Attempt to generate the tool state using the standard Galaxy-side code
return tool.new_state(trans)
except Exception as e:
# Fall back to building tool state as below
log.debug(
'Failed to build tool state for tool "%s" using standard method, will try to fall back on custom method: %s',
tool.id,
e,
)
inputs = tool.inputs_by_page[0]
context = ExpressionContext(state.inputs, parent=None)
for input in inputs.values():
try:
state.inputs[input.name] = input.get_initial_value(trans, context)
except Exception:
# FIXME: not all values should be an empty list
state.inputs[input.name] = []
return state
__all__ = (
"copy_sample_file",
lib/tool_shed/webapp/api/categories.py
+2
View File
@@ -1,5 +1,6 @@
import logging
from typing import (
Any,
Callable,
Dict,
)
@@ -84,6 +85,7 @@ class CategoriesController(BaseAPIController):
sort_order = kwd.get("sort_order", "asc")
page = kwd.get("page", None)
category = suc.get_category(self.app, category_id)
category_dict: Dict[str, Any]
if category is None:
category_dict = dict(message=f"Unable to locate category record for id {str(id)}.", status="error")
return category_dict
lib/tool_shed/webapp/api/repositories.py
+1 -1
View File
@@ -910,7 +910,7 @@ class RepositoriesController(BaseAPIController):
category_ids=category_ids,
)
repo, message = repository_util.update_repository(app=self.app, trans=trans, id=id, **update_kwds)
repo, message = repository_util.update_repository(trans, id, **update_kwds)
if repo is None:
if "You are not the owner" in message:
raise InsufficientPermissionsException(message)
lib/tool_shed/webapp/controllers/repository.py
+1 -1
View File
@@ -1698,7 +1698,7 @@ class RepositoryController(BaseUIController, ratings_util.ItemRatings):
type=repository_type,
)
repository, message = repository_util.update_repository(app=trans.app, trans=trans, id=id, **update_kwds)
repository, message = repository_util.update_repository(trans=trans, id=id, **update_kwds)
if repository is None:
return trans.response.send_redirect(
web.url_for(
lib/tool_shed/webapp/templates/webapps/tool_shed/repository/common.mako
-223
View File
@@ -397,32 +397,6 @@
%>
</%def>
<%def name="render_failed_test( failed_test, pad, parent, row_counter, row_is_header=False, render_repository_actions_for='tool_shed' )">
<%
from tool_shed.util.basic_util import to_html_string
encoded_id = trans.security.encode_id( failed_test.id )
%>
<tr class="datasetRow"
%if parent is not None:
parent="${parent}"
%endif
id="libraryItem-rft-${encoded_id}">
<td style="padding-left: ${pad+20}px;">
<table id="test_environment">
<tr><td bgcolor="#FFFFCC"><b>Tool id:</b> ${failed_test.tool_id | h}</td></tr>
<tr><td><b>Tool version:</b> ${failed_test.tool_id | h}</td></tr>
<tr><td><b>Test:</b> ${failed_test.test_id | h}</td></tr>
<tr><td><b>Stderr:</b> <br/>${ to_html_string( failed_test.stderr ) }</td></tr>
<tr><td><b>Traceback:</b> <br/>${ to_html_string( failed_test.traceback ) }</td></tr>
</table>
</td>
</tr>
<%
my_row = row_counter.count
row_counter.increment()
%>
</%def>
<%def name="render_invalid_data_manager( data_manager, pad, parent, row_counter, row_is_header=False, render_repository_actions_for='tool_shed' )">
<%
encoded_id = trans.security.encode_id( data_manager.id )
@@ -648,176 +622,6 @@
</style>
</%def>
<%def name="render_tool_dependency_installation_error( installation_error, pad, parent, row_counter, row_is_header=False, render_repository_actions_for='tool_shed' )">
<%
from galaxy.util import unicodify
encoded_id = trans.security.encode_id( installation_error.id )
%>
<tr class="datasetRow"
%if parent is not None:
parent="${parent}"
%endif
id="libraryItem-rtdie-${encoded_id}">
<td style="padding-left: ${pad+20}px;">
<table id="test_environment">
<tr bgcolor="#FFFFCC">
<th>Type</th><th>Name</th><th>Version</th>
</tr>
<tr>
<td>${installation_error.name | h}</td>
<td>${installation_error.type | h}</td>
<td>${installation_error.version | h}</td>
</tr>
<tr><th>Error</th></tr>
<tr><td colspan="3">${unicodify( installation_error.error_message ) | h}</td></tr>
</table>
</td>
</tr>
<%
my_row = row_counter.count
row_counter.increment()
%>
</%def>
<%def name="render_tool_dependency_successful_installation( successful_installation, pad, parent, row_counter, row_is_header=False, render_repository_actions_for='tool_shed' )">
<%
encoded_id = trans.security.encode_id( successful_installation.id )
%>
<tr class="datasetRow"
%if parent is not None:
parent="${parent}"
%endif
id="libraryItem-rtdsi-${encoded_id}">
<td style="padding-left: ${pad+20}px;">
<table id="test_environment">
<tr bgcolor="#FFFFCC">
<th>Type</th><th>Name</th><th>Version</th>
</tr>
<tr>
<td>${successful_installation.name | h}</td>
<td>${successful_installation.type | h}</td>
<td>${successful_installation.version | h}</td>
</tr>
<tr><th>Installation directory</th></tr>
<tr><td colspan="3">${successful_installation.installation_directory | h}</td></tr>
</table>
</td>
</tr>
<%
my_row = row_counter.count
row_counter.increment()
%>
</%def>
<%def name="render_repository_installation_error( installation_error, pad, parent, row_counter, row_is_header=False, is_current_repository=False, render_repository_actions_for='tool_shed' )">
<%
from galaxy.util import unicodify
encoded_id = trans.security.encode_id( installation_error.id )
%>
<tr class="datasetRow"
%if parent is not None:
parent="${parent}"
%endif
id="libraryItem-rrie-${encoded_id}">
<td style="padding-left: ${pad+20}px;">
<table id="test_environment">
%if not is_current_repository:
<tr bgcolor="#FFFFCC">
<th>Tool shed</th><th>Name</th><th>Owner</th><th>Changeset revision</th>
</tr>
<tr>
<td>${installation_error.tool_shed | h}</td>
<td>${installation_error.name | h}</td>
<td>${installation_error.owner | h}</td>
<td>${installation_error.changeset_revision | h}</td>
</tr>
%endif
<tr><th>Error</th></tr>
<tr><td colspan="4">${unicodify( installation_error.error_message ) | h}</td></tr>
</table>
</td>
</tr>
<%
my_row = row_counter.count
row_counter.increment()
%>
</%def>
<%def name="render_repository_successful_installation( successful_installation, pad, parent, row_counter, row_is_header=False, is_current_repository=False, render_repository_actions_for='tool_shed' )">
<%
encoded_id = trans.security.encode_id( successful_installation.id )
%>
<tr class="datasetRow"
%if parent is not None:
parent="${parent}"
%endif
id="libraryItem-rrsi-${encoded_id}">
<td style="padding-left: ${pad+20}px;">
<table id="test_environment">
%if not is_current_repository:
<tr bgcolor="#FFFFCC">
<th>Tool shed</th><th>Name</th><th>Owner</th><th>Changeset revision</th>
</tr>
<tr>
<td>${successful_installation.tool_shed | h}</td>
<td>${successful_installation.name | h}</td>
<td>${successful_installation.owner | h}</td>
<td>${successful_installation.changeset_revision | h}</td>
</tr>
%endif
</table>
</td>
</tr>
<%
my_row = row_counter.count
row_counter.increment()
%>
</%def>
<%def name="render_not_tested( not_tested, pad, parent, row_counter, row_is_header=False, render_repository_actions_for='tool_shed' )">
<%
encoded_id = trans.security.encode_id( not_tested.id )
%>
<tr class="datasetRow"
%if parent is not None:
parent="${parent}"
%endif
id="libraryItem-rnt-${encoded_id}">
<td style="padding-left: ${pad+20}px;">
<table id="test_environment">
<tr><td>${not_tested.reason | h}</td></tr>
</table>
</td>
</tr>
<%
my_row = row_counter.count
row_counter.increment()
%>
</%def>
<%def name="render_passed_test( passed_test, pad, parent, row_counter, row_is_header=False, render_repository_actions_for='tool_shed' )">
<%
encoded_id = trans.security.encode_id( passed_test.id )
%>
<tr class="datasetRow"
%if parent is not None:
parent="${parent}"
%endif
id="libraryItem-rpt-${encoded_id}">
<td style="padding-left: ${pad+20}px;">
<table id="test_environment">
<tr><td bgcolor="#FFFFCC"><b>Tool id:</b> ${passed_test.tool_id | h}</td></tr>
<tr><td><b>Tool version:</b> ${passed_test.tool_id | h}</td></tr>
<tr><td><b>Test:</b> ${passed_test.test_id | h}</td></tr>
</table>
</td>
</tr>
<%
my_row = row_counter.count
row_counter.increment()
%>
</%def>
<%def name="render_tool( tool, pad, parent, row_counter, row_is_header, render_repository_actions_for='tool_shed' )">
<%
encoded_id = trans.security.encode_id( tool.id )
@@ -924,33 +728,6 @@
%>
</%def>
<%def name="render_test_environment( test_environment, pad, parent, row_counter, row_is_header=False, render_repository_actions_for='tool_shed' )">
<% encoded_id = trans.security.encode_id( test_environment.id ) %>
<tr class="datasetRow"
%if parent is not None:
parent="${parent}"
%endif
id="libraryItem-rte-${encoded_id}">
<td style="padding-left: ${pad+20}px;">
<table class="grid" id="test_environment">
<tr><td><b>Time tested:</b> ${test_environment.time_tested | h}</td></tr>
<tr><td><b>System:</b> ${test_environment.system | h}</td></tr>
<tr><td><b>Architecture:</b> ${test_environment.architecture | h}</td></tr>
<tr><td><b>Python version:</b> ${test_environment.python_version | h}</td></tr>
<tr><td><b>Galaxy revision:</b> ${test_environment.galaxy_revision | h}</td></tr>
<tr><td><b>Galaxy database version:</b> ${test_environment.galaxy_database_version | h}</td></tr>
<tr><td><b>Tool shed revision:</b> ${test_environment.tool_shed_revision | h}</td></tr>
<tr><td><b>Tool shed database version:</b> ${test_environment.tool_shed_database_version | h}</td></tr>
<tr><td><b>Tool shed mercurial version:</b> ${test_environment.tool_shed_mercurial_version | h}</td></tr>
</table>
</td>
</tr>
<%
my_row = row_counter.count
row_counter.increment()
%>
</%def>
<%def name="render_valid_data_manager( data_manager, pad, parent, row_counter, row_is_header=False, render_repository_actions_for='tool_shed' )">
<%
encoded_id = trans.security.encode_id( data_manager.id )
lib/tool_shed_client/schema/__init__.py
+219 -2
View File
@@ -1,12 +1,21 @@
from typing import (
Any,
cast,
Dict,
List,
Optional,
Tuple,
Union,
)
from pydantic import BaseModel
from pydantic import (
BaseModel,
Field,
)
from typing_extensions import (
Literal,
TypedDict,
)
class Repository(BaseModel):
@@ -26,6 +35,17 @@ class Repository(BaseModel):
create_time: str
class CreateUserRequest(BaseModel):
username: str
email: str
password: str
class User(BaseModel):
id: str
username: str
class Category(BaseModel):
id: str
name: str
@@ -33,7 +53,7 @@ class Category(BaseModel):
class CreateCategoryRequest(BaseModel):
name: str
description: str
description: Optional[str] = None
class ValidRepostiroyUpdateMessage(BaseModel):
@@ -53,6 +73,38 @@ class OrderedInstallableRevisions(BaseModel):
__root__: List[str]
RepositoryType = Literal[
"repository_suite_definition",
"tool_dependency_definition",
"unrestricted",
]
class CreateRepositoryRequest(BaseModel):
name: str
synopsis: str
description: Optional[str] = None
remote_repository_url: Optional[str] = None
homepage_url: Optional[str] = None
type_: RepositoryType = Field(
"unrestricted",
alias="type",
title="Type",
)
category_ids: str = Field(
...,
alias="category_ids[]",
title="Category IDs",
)
class Config:
allow_population_by_field_name = True
class RepositoryUpdateRequest(BaseModel):
commit_message: Optional[str] = None
class RepositoryUpdate(BaseModel):
__root__: Union[ValidRepostiroyUpdateMessage, FailedRepositoryUpdateMessage]
@@ -203,3 +255,168 @@ class RepositorySearchResults(BaseModel):
page_size: str
hostname: str
hits: List[RepositorySearchHit]
class GetInstallInfoRequest(BaseModel):
owner: str
name: str
changeset_revision: str
class ValidToolDict(TypedDict):
add_to_tool_panel: bool
description: str
guid: str
id: str
name: str
requirements: list
tests: list
tool_config: str
tool_type: str
version: str
version_string_cmd: Optional[str]
class RepositoryMetadataInstallInfoDict(TypedDict):
changeset_revision: str
downloadable: bool
has_repository_dependencies: bool
has_repository_dependencies_only_if_compiling_contained_td: bool
id: str
includes_datatypes: bool
includes_tool_dependencies: bool
includes_tools: bool
includes_tools_for_display_in_tool_panel: bool
includes_workflows: bool
malicious: bool
repository_id: str
url: str
valid_tools: List[ValidToolDict]
# So hard to type this... the keys are repo names and the elements
# are tuples that have been list-ified.
ExtraRepoInfo = Dict[str, List]
# {
# "add_column": [
# "add_column hello",
# "http://test@localhost:9009/repos/test/add_column",
# "3a08cc21466f",
# "1",
# "test",
# {},
# {}
# ]
# }
EmptyDict = TypedDict("EmptyDict", {})
LegacyInstallInfoTuple = Tuple[
Optional[Dict], Union[RepositoryMetadataInstallInfoDict, EmptyDict], Union[ExtraRepoInfo, EmptyDict]
]
class RepositoryExtraInstallInfo(BaseModel):
name: str
description: str
repository_clone_url: str
changeset_revision: str
ctx_rev: str
repository_owner: str
repository_dependencies: Optional[Dict]
# tool dependencies not longer work so don't transmit them in v2?
# tool_dependencies: Optional[Dict]
@staticmethod
def from_legacy_dict(as_dict: ExtraRepoInfo) -> "RepositoryExtraInstallInfo":
assert len(as_dict) == 1
repo_name = next(iter(as_dict.keys()))
info_indexable = as_dict[repo_name]
return RepositoryExtraInstallInfo(
name=repo_name,
description=info_indexable[0],
repository_clone_url=info_indexable[1],
changeset_revision=info_indexable[2],
ctx_rev=info_indexable[3],
repository_owner=info_indexable[4],
repository_dependencies=info_indexable[5],
)
class ValidTool(BaseModel):
add_to_tool_panel: bool
description: str
guid: str
id: str
name: str
requirements: list
tests: list
tool_config: str
tool_type: str
version: str
version_string_cmd: Optional[str]
@staticmethod
def from_legacy_dict(as_dict: ValidToolDict) -> "ValidTool":
return ValidTool(**as_dict)
@staticmethod
def from_legacy_list(as_dicts: List[ValidToolDict]) -> List["ValidTool"]:
return list(ValidTool.from_legacy_dict(d) for d in as_dicts)
class RepositoryMetadataInstallInfo(BaseModel):
id: str
changeset_revision: str
downloadable: bool
has_repository_dependencies: bool
includes_tools: bool
includes_tools_for_display_in_tool_panel: bool
malicious: bool
repository_id: str
url: str
valid_tools: List[ValidToolDict]
# no longer used, don't transmit.
# has_repository_dependencies_only_if_compiling_contained_td: bool
# includes_datatypes: bool
# includes_tool_dependencies: bool
# includes_workflows: bool
@staticmethod
def from_legacy_dict(as_dict: RepositoryMetadataInstallInfoDict) -> "RepositoryMetadataInstallInfo":
return RepositoryMetadataInstallInfo(
id=as_dict["id"],
changeset_revision=as_dict["changeset_revision"],
downloadable=as_dict["downloadable"],
has_repository_dependencies=as_dict["has_repository_dependencies"],
includes_tools=as_dict["includes_tools"],
includes_tools_for_display_in_tool_panel=as_dict["includes_tools_for_display_in_tool_panel"],
malicious=as_dict["malicious"],
repository_id=as_dict["repository_id"],
url=as_dict["url"],
valid_tools=ValidTool.from_legacy_list(as_dict["valid_tools"]),
)
class InstallInfo(BaseModel):
metadata_info: Optional[RepositoryMetadataInstallInfo]
repo_info: Optional[RepositoryExtraInstallInfo]
def from_legacy_install_info(legacy_install_info: LegacyInstallInfoTuple) -> InstallInfo:
repo_metadata_install_info: Union[RepositoryMetadataInstallInfoDict, EmptyDict]
extra_info: Union[ExtraRepoInfo, EmptyDict]
_, repo_metadata_install_info, extra_info = legacy_install_info
if repo_metadata_install_info:
metadata_info = RepositoryMetadataInstallInfo.from_legacy_dict(
cast(RepositoryMetadataInstallInfoDict, repo_metadata_install_info)
)
else:
metadata_info = None
if extra_info:
repo_info = RepositoryExtraInstallInfo.from_legacy_dict(extra_info)
else:
repo_info = None
return InstallInfo(
metadata_info=metadata_info,
repo_info=repo_info,
)