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

Enhance schema models.

Browse Source
This commit is contained in:
John Chilton
2023-02-03 13:47:38 -05:00
parent 18f73b5532
commit 6b3a2d3cd6
5 changed files with 281 additions and 28 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/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/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_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,
)