import json
import logging
import os
from io import StringIO
from time import strftime
from typing import (
Callable,
Dict,
)
from webob.compat import cgi_FieldStorage
from galaxy import (
util,
web,
)
from galaxy.exceptions import (
ActionInputError,
InsufficientPermissionsException,
MessageException,
ObjectNotFound,
RequestParameterInvalidException,
RequestParameterMissingException,
)
from galaxy.web import (
expose_api,
expose_api_anonymous_and_sessionless,
expose_api_raw_anonymous_and_sessionless,
)
from galaxy.webapps.base.controller import HTTPBadRequest
from tool_shed.managers.repositories import (
can_update_repo,
check_updates,
create_repository,
get_install_info,
get_ordered_installable_revisions,
get_repository_metadata_dict,
get_value_mapper,
index_repositories,
index_tool_ids,
reset_metadata_on_repository,
search,
to_element_dict,
UpdatesRequest,
upload_tar_and_set_metadata,
)
from tool_shed.metadata import repository_metadata_manager
from tool_shed.repository_types import util as rt_util
from tool_shed.util import (
metadata_util,
repository_util,
tool_util,
)
from tool_shed.webapp import model
from tool_shed_client.schema import (
CreateRepositoryRequest,
LegacyInstallInfoTuple,
)
from . import BaseShedAPIController
log = logging.getLogger(__name__)
[docs]class RepositoriesController(BaseShedAPIController):
"""RESTful controller for interactions with repositories in the Tool Shed."""
[docs] @web.legacy_expose_api
def add_repository_registry_entry(self, trans, payload, **kwd):
"""
POST /api/repositories/add_repository_registry_entry
Adds appropriate entries to the repository registry for the repository defined by the received name and owner.
:param key: the user's API key
The following parameters are included in the payload.
:param tool_shed_url (required): the base URL of the Tool Shed containing the Repository
:param name (required): the name of the Repository
:param owner (required): the owner of the Repository
"""
response_dict = {}
if not trans.user_is_admin:
response_dict["status"] = "error"
response_dict["message"] = "You are not authorized to add entries to this Tool Shed's repository registry."
return response_dict
tool_shed_url = payload.get("tool_shed_url", "")
if not tool_shed_url:
raise HTTPBadRequest(detail="Missing required parameter 'tool_shed_url'.")
tool_shed_url = tool_shed_url.rstrip("/")
name = payload.get("name", "")
if not name:
raise HTTPBadRequest(detail="Missing required parameter 'name'.")
owner = payload.get("owner", "")
if not owner:
raise HTTPBadRequest(detail="Missing required parameter 'owner'.")
repository = repository_util.get_repository_by_name_and_owner(self.app, name, owner)
if repository is None:
error_message = f"Cannot locate repository with name {name} and owner {owner},"
log.debug(error_message)
response_dict["status"] = "error"
response_dict["message"] = error_message
return response_dict
# Update the repository registry.
self.app.repository_registry.add_entry(repository)
response_dict["status"] = "ok"
response_dict["message"] = (
f"Entries for repository {name} owned by {owner} have been added to the Tool Shed repository registry."
)
return response_dict
[docs] @web.legacy_expose_api_anonymous
def get_ordered_installable_revisions(self, trans, name=None, owner=None, **kwd):
"""
GET /api/repositories/get_ordered_installable_revisions
:param name: the name of the Repository
:param owner: the owner of the Repository
Returns the ordered list of changeset revision hash strings that are associated with installable revisions.
As in the changelog, the list is ordered oldest to newest.
"""
# Example URL: http://localhost:9009/api/repositories/get_ordered_installable_revisions?name=add_column&owner=test
if name is None:
name = kwd.get("name", None)
if owner is None:
owner = kwd.get("owner", None)
tsr_id = kwd.get("tsr_id", None)
return get_ordered_installable_revisions(self.app, name, owner, tsr_id)
[docs] @web.legacy_expose_api_anonymous
def get_repository_revision_install_info(
self, trans, name, owner, changeset_revision, **kwd
) -> LegacyInstallInfoTuple:
"""
GET /api/repositories/get_repository_revision_install_info
:param name: the name of the Repository
:param owner: the owner of the Repository
:param changeset_revision: the changeset_revision of the RepositoryMetadata object associated with the Repository
Returns a list of the following dictionaries
- a dictionary defining the Repository. For example::
{
"deleted": false,
"deprecated": false,
"description": "add_column hello",
"id": "f9cad7b01a472135",
"long_description": "add_column hello",
"name": "add_column",
"owner": "test",
"private": false,
"times_downloaded": 6,
"url": "/api/repositories/f9cad7b01a472135",
"user_id": "f9cad7b01a472135"
}
- a dictionary defining the Repository revision (RepositoryMetadata). For example::
{
"changeset_revision": "3a08cc21466f",
"downloadable": true,
"has_repository_dependencies": false,
"has_repository_dependencies_only_if_compiling_contained_td": false,
"id": "f9cad7b01a472135",
"includes_datatypes": false,
"includes_tool_dependencies": false,
"includes_tools": true,
"includes_tools_for_display_in_tool_panel": true,
"includes_workflows": false,
"malicious": false,
"repository_id": "f9cad7b01a472135",
"url": "/api/repository_revisions/f9cad7b01a472135",
"valid_tools": [{u'add_to_tool_panel': True,
u'description': u'data on any column using simple expressions',
u'guid': u'localhost:9009/repos/enis/sample_repo_1/Filter1/2.2.0',
u'id': u'Filter1',
u'name': u'Filter',
u'requirements': [],
u'tests': [{u'inputs': [[u'input', u'1.bed'], [u'cond', u"c1=='chr22'"]],
u'name': u'Test-1',
u'outputs': [[u'out_file1', u'filter1_test1.bed']],
u'required_files': [u'1.bed', u'filter1_test1.bed']}],
u'tool_config': u'database/community_files/000/repo_1/filtering.xml',
u'tool_type': u'default',
u'version': u'2.2.0',
u'version_string_cmd': None}]
}
- a dictionary including the additional information required to install the repository. For example::
{
"add_column": [
"add_column hello",
"http://test@localhost:9009/repos/test/add_column",
"3a08cc21466f",
"1",
"test",
{},
{}
]
}
"""
return get_install_info(trans, name, owner, changeset_revision)
[docs] @web.legacy_expose_api_anonymous
def get_installable_revisions(self, trans, **kwd):
"""
GET /api/repositories/get_installable_revisions
:param tsr_id: the encoded toolshed ID of the repository
Returns a list of lists of changesets, in the format [ [ 0, fbb391dc803c ], [ 1, 9d9ec4d9c03e ], [ 2, 9b5b20673b89 ], [ 3, e8c99ce51292 ] ].
"""
# Example URL: http://localhost:9009/api/repositories/get_installable_revisions?tsr_id=9d37e53072ff9fa4
if (tsr_id := kwd.get("tsr_id", None)) is not None:
repository = repository_util.get_repository_in_tool_shed(
self.app, tsr_id, eagerload_columns=[model.Repository.downloadable_revisions]
)
else:
error_message = "Error in the Tool Shed repositories API in get_ordered_installable_revisions: "
error_message += "missing or invalid parameter received."
log.debug(error_message)
return []
return repository.installable_revisions(self.app)
def __get_value_mapper(self, trans) -> Dict[str, Callable]:
return get_value_mapper(self.app)
[docs] @expose_api_raw_anonymous_and_sessionless
def index(self, trans, deleted=False, owner=None, name=None, **kwd):
"""
GET /api/repositories
Displays a collection of repositories with optional criteria.
:param q: (optional)if present search on the given query will be performed
:type q: str
:param page: (optional)requested page of the search
:type page: int
:param page_size: (optional)requested page_size of the search
:type page_size: int
:param jsonp: (optional)flag whether to use jsonp format response, defaults to False
:type jsonp: bool
:param callback: (optional)name of the function to wrap callback in
used only when jsonp is true, defaults to 'callback'
:type callback: str
:param deleted: (optional)displays repositories that are or are not set to deleted.
:type deleted: bool
:param owner: (optional)the owner's public username.
:type owner: str
:param name: (optional)the repository name.
:type name: str
:param tool_ids: (optional) a tool GUID to find the repository for
:param tool_ids: str
:returns dict: object containing list of results
Examples:
GET http://localhost:9009/api/repositories
GET http://localhost:9009/api/repositories?q=fastq
"""
repository_dicts = []
deleted = util.asbool(deleted)
if q := kwd.get("q", ""):
page = kwd.get("page", 1)
page_size = kwd.get("page_size", 10)
try:
page = int(page)
page_size = int(page_size)
except ValueError:
raise RequestParameterInvalidException('The "page" and "page_size" parameters have to be integers.')
return_jsonp = util.asbool(kwd.get("jsonp", False))
callback = kwd.get("callback", "callback")
search_results = search(trans, q, page, page_size)
if return_jsonp:
response = str(f"{callback}({json.dumps(search_results)});")
else:
response = json.dumps(search_results)
return response
if (tool_ids := kwd.get("tool_ids", None)) is not None:
tool_ids = util.listify(tool_ids)
response = index_tool_ids(self.app, tool_ids)
return json.dumps(response)
else:
repositories = index_repositories(self.app, name, owner, deleted)
repository_dicts = []
for repository in repositories:
repository_dict = repository.to_dict(view="collection", value_mapper=self.__get_value_mapper(trans))
repository_dict["category_ids"] = [
trans.security.encode_id(x.category.id) for x in repository.categories
]
repository_dicts.append(repository_dict)
return json.dumps(repository_dicts)
[docs] @web.legacy_expose_api
def remove_repository_registry_entry(self, trans, payload, **kwd):
"""
POST /api/repositories/remove_repository_registry_entry
Removes appropriate entries from the repository registry for the repository defined by the received name and owner.
:param key: the user's API key
The following parameters are included in the payload.
:param tool_shed_url (required): the base URL of the Tool Shed containing the Repository
:param name (required): the name of the Repository
:param owner (required): the owner of the Repository
"""
response_dict = {}
if not trans.user_is_admin:
response_dict["status"] = "error"
response_dict["message"] = (
"You are not authorized to remove entries from this Tool Shed's repository registry."
)
return response_dict
tool_shed_url = payload.get("tool_shed_url", "")
if not tool_shed_url:
raise HTTPBadRequest(detail="Missing required parameter 'tool_shed_url'.")
tool_shed_url = tool_shed_url.rstrip("/")
name = payload.get("name", "")
if not name:
raise HTTPBadRequest(detail="Missing required parameter 'name'.")
owner = payload.get("owner", "")
if not owner:
raise HTTPBadRequest(detail="Missing required parameter 'owner'.")
repository = repository_util.get_repository_by_name_and_owner(self.app, name, owner)
if repository is None:
error_message = f"Cannot locate repository with name {name} and owner {owner},"
log.debug(error_message)
response_dict["status"] = "error"
response_dict["message"] = error_message
return response_dict
# Update the repository registry.
self.app.repository_registry.remove_entry(repository)
response_dict["status"] = "ok"
response_dict["message"] = (
f"Entries for repository {name} owned by {owner} have been removed from the Tool Shed repository registry."
)
return response_dict
[docs] @expose_api_anonymous_and_sessionless
def show(self, trans, id, **kwd):
"""
GET /api/repositories/{encoded_repository_id}
Returns information about a repository in the Tool Shed.
Example URL: http://localhost:9009/api/repositories/f9cad7b01a472135
:param id: the encoded id of the Repository object
:type id: encoded str
:returns: detailed repository information
:rtype: dict
:raises: ObjectNotFound, MalformedId
"""
repository = repository_util.get_repository_in_tool_shed(self.app, id)
if repository is None:
raise ObjectNotFound("Unable to locate repository for the given id.")
repository_dict = repository.to_dict(view="element", value_mapper=self.__get_value_mapper(trans))
# TODO the following property would be better suited in the to_dict method
repository_dict["category_ids"] = [trans.security.encode_id(x.category.id) for x in repository.categories]
return repository_dict
[docs] @expose_api_raw_anonymous_and_sessionless
def updates(self, trans, **kwd):
"""
GET /api/repositories/updates
Return a dictionary with boolean values for whether there are updates available
for the repository revision, newer installable revisions available,
the revision is the latest installable revision, and if the repository is deprecated.
:param owner: owner of the repository
:type owner: str
:param name: name of the repository
:type name: str
:param changeset_revision: changeset of the repository
:type changeset_revision: str
:param hexlify: flag whether to hexlify the response (for backward compatibility)
:type changeset: boolean
:returns: information about repository deprecations, updates, and upgrades
:rtype: dict
"""
name = kwd.get("name", None)
owner = kwd.get("owner", None)
changeset_revision = kwd.get("changeset_revision", None)
hexlify_this = util.asbool(kwd.get("hexlify", True))
request = UpdatesRequest(
name=name,
owner=owner,
changeset_revision=changeset_revision,
hexlify=hexlify_this,
)
return check_updates(trans.app, request)
[docs] @expose_api
def update(self, trans, id, **kwd):
"""
PATCH /api/repositories/{encoded_repository_id}
Updates information about a repository in the Tool Shed.
:param id: the encoded id of the Repository object
:param payload: dictionary structure containing
'name': repo's name (optional)
'synopsis': repo's synopsis (optional)
'description': repo's description (optional)
'remote_repository_url': repo's remote repo (optional)
'homepage_url': repo's homepage url (optional)
'category_ids': list of existing encoded TS category ids the updated repo should be associated with (optional)
:type payload: dict
:returns: detailed repository information
:rtype: dict
:raises: RequestParameterInvalidException, InsufficientPermissionsException
"""
payload = kwd.get("payload", None)
if not payload:
raise RequestParameterMissingException("You did not specify any payload.")
name = payload.get("name", None)
synopsis = payload.get("synopsis", None)
description = payload.get("description", None)
remote_repository_url = payload.get("remote_repository_url", None)
homepage_url = payload.get("homepage_url", None)
category_ids = payload.get("category_ids", None)
if category_ids is not None:
# We need to know if it was actually passed, and listify turns None into []
category_ids = util.listify(category_ids)
update_kwds = dict(
name=name,
description=synopsis,
long_description=description,
remote_repository_url=remote_repository_url,
homepage_url=homepage_url,
category_ids=category_ids,
)
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)
else:
raise ActionInputError(message)
repository_dict = repo.to_dict(view="element", value_mapper=self.__get_value_mapper(trans))
repository_dict["category_ids"] = [trans.security.encode_id(x.category.id) for x in repo.categories]
return repository_dict
[docs] @expose_api
def create(self, trans, **kwd):
"""
POST /api/repositories:
Creates a new repository.
Only ``name`` and ``synopsis`` parameters are required.
:param payload: dictionary structure containing
'name': new repo's name (required)
'synopsis': new repo's synopsis (required)
'description': new repo's description (optional)
'remote_repository_url': new repo's remote repo (optional)
'homepage_url': new repo's homepage url (optional)
'category_ids[]': list of existing encoded TS category ids the new repo should be associated with (optional)
'type': new repo's type, defaults to ``unrestricted`` (optional)
:type payload: dict
:returns: detailed repository information
:rtype: dict
:raises: RequestParameterMissingException, RequestParameterInvalidException
"""
payload = kwd.get("payload", None)
if not payload:
raise RequestParameterMissingException("You did not specify any payload.")
name = payload.get("name", None)
if not name:
raise RequestParameterMissingException("Missing required parameter 'name'.")
synopsis = payload.get("synopsis", None)
if not synopsis:
raise RequestParameterMissingException("Missing required parameter 'synopsis'.")
description = payload.get("description", "")
remote_repository_url = payload.get("remote_repository_url", "")
homepage_url = payload.get("homepage_url", "")
repo_type = payload.get("type", rt_util.UNRESTRICTED)
if repo_type not in rt_util.types:
raise RequestParameterInvalidException("This repository type is not valid")
request = CreateRepositoryRequest(
name=name,
synopsis=synopsis,
description=description,
remote_repository_url=remote_repository_url,
homepage_url=homepage_url,
category_ids=payload.get("category_ids[]", ""),
type_=repo_type,
)
repo = create_repository(trans, request)
return to_element_dict(self.app, repo, include_categories=True)
[docs] @web.legacy_expose_api
def create_changeset_revision(self, trans, id, payload, **kwd):
"""
POST /api/repositories/{encoded_repository_id}/changeset_revision
Create a new tool shed repository commit - leaving PUT on parent
resource open for updating meta-attributes of the repository (and
Galaxy doesn't allow PUT multipart data anyway
https://trello.com/c/CQwmCeG6).
:param id: the encoded id of the Repository object
The following parameters may be included in the payload.
:param commit_message: hg commit message for update.
"""
# Example URL: http://localhost:9009/api/repositories/f9cad7b01a472135
repository = repository_util.get_repository_in_tool_shed(self.app, id)
if not can_update_repo(trans, repository):
trans.response.status = 400
return {
"err_msg": "You do not have permission to update this repository.",
}
file_data = payload.get("file")
# Code stolen from gx's upload_common.py
if isinstance(file_data, cgi_FieldStorage):
assert not isinstance(file_data.file, StringIO)
assert file_data.file.name != "<fdopen>"
local_filename = util.mkstemp_ln(file_data.file.name, "upload_file_data_")
file_data.file.close()
file_data = dict(filename=file_data.filename, local_filename=local_filename)
elif isinstance(file_data, dict) and "local_filename" not in file_data:
raise Exception("Uploaded file was encoded in a way not understood.")
commit_message = kwd.get("commit_message", "Uploaded")
uploaded_file_name = file_data["local_filename"]
try:
message = upload_tar_and_set_metadata(
trans,
trans.request.host,
repository,
uploaded_file_name,
commit_message,
)
rval = {"message": message}
except MessageException as e:
trans.response.status = e.status_code
rval = {"err_msg": str(e)}
if os.path.exists(uploaded_file_name):
os.remove(uploaded_file_name)
return rval