import logging
import re
from galaxy import util
from galaxy.model.base import transaction
from galaxy.tool_shed.util import repository_util
from galaxy.util.tool_shed import common_util
log = logging.getLogger(__name__)
MAX_CONTENT_SIZE = 1048576
DATATYPES_CONFIG_FILENAME = "datatypes_conf.xml"
REPOSITORY_DATA_MANAGER_CONFIG_FILENAME = "data_manager_conf.xml"
[docs]def can_eliminate_repository_dependency(metadata_dict, tool_shed_url, name, owner):
"""
Determine if the relationship between a repository_dependency record
associated with a tool_shed_repository record on the Galaxy side
can be eliminated.
"""
rd_dict = metadata_dict.get("repository_dependencies", {})
rd_tups = rd_dict.get("repository_dependencies", [])
for rd_tup in rd_tups:
tsu, n, o, none1, none2, none3 = common_util.parse_repository_dependency_tuple(rd_tup)
if tsu == tool_shed_url and n == name and o == owner:
# The repository dependency is current, so keep it.
return False
return True
[docs]def clean_dependency_relationships(trans, metadata_dict, tool_shed_repository, tool_shed_url):
"""
Repositories of type tool_dependency_definition allow for defining a
package dependency at some point in the change log and then removing the
dependency later in the change log. This function keeps the dependency
relationships on the Galaxy side current by deleting database records
that defined the now-broken relationships.
"""
for rrda in tool_shed_repository.required_repositories:
rd = rrda.repository_dependency
r = rd.repository
if can_eliminate_repository_dependency(metadata_dict, tool_shed_url, r.name, r.owner):
message = "Repository dependency %s by owner %s is not required by repository %s, owner %s, "
message += "removing from list of repository dependencies."
log.debug(message % (r.name, r.owner, tool_shed_repository.name, tool_shed_repository.owner))
session = trans.install_model.context
session.delete(rrda)
with transaction(session):
session.commit()
[docs]def get_ctx_rev(app, tool_shed_url, name, owner, changeset_revision):
"""
Send a request to the tool shed to retrieve the ctx_rev for a repository defined by the
combination of a name, owner and changeset revision.
"""
tool_shed_url = common_util.get_tool_shed_url_from_tool_shed_registry(app, tool_shed_url)
params = dict(name=name, owner=owner, changeset_revision=changeset_revision)
pathspec = ["repository", "get_ctx_rev"]
ctx_rev = util.url_get(
tool_shed_url, auth=app.tool_shed_registry.url_auth(tool_shed_url), pathspec=pathspec, params=params
)
return ctx_rev
[docs]def get_next_prior_import_or_install_required_dict_entry(prior_required_dict, processed_tsr_ids):
"""
This method is used in the Tool Shed when exporting a repository and its dependencies, and in Galaxy
when a repository and its dependencies are being installed. The order in which the prior_required_dict
is processed is critical in order to ensure that the ultimate repository import or installation order is
correctly defined. This method determines the next key / value pair from the received prior_required_dict
that should be processed.
"""
# Return the first key / value pair that is not yet processed and whose value is an empty list.
for key, value in prior_required_dict.items():
if key in processed_tsr_ids:
continue
if not value:
return key
# Return the first key / value pair that is not yet processed and whose ids in value are all included
# in processed_tsr_ids.
for key, value in prior_required_dict.items():
if key in processed_tsr_ids:
continue
all_contained = True
for required_repository_id in value:
if required_repository_id not in processed_tsr_ids:
all_contained = False
break
if all_contained:
return key
# Return the first key / value pair that is not yet processed. Hopefully this is all that is necessary
# at this point.
for key in prior_required_dict.keys():
if key in processed_tsr_ids:
continue
return key
[docs]def get_user(app, id):
"""Get a user from the database by id."""
sa_session = app.model.session
return sa_session.query(app.model.User).get(app.security.decode_id(id))
[docs]def set_image_paths(app, text, encoded_repository_id=None, tool_shed_repository=None, tool_id=None, tool_version=None):
"""
Handle tool help image display for tools that are contained in repositories in
the tool shed or installed into Galaxy as well as image display in repository
README files. This method will determine the location of the image file and
return the path to it that will enable the caller to open the file.
"""
if text:
if repository_util.is_tool_shed_client(app) and encoded_repository_id:
route_to_images = f"admin_toolshed/static/images/{encoded_repository_id}"
elif encoded_repository_id:
# We're in the tool shed.
route_to_images = f"/repository/static/images/{encoded_repository_id}"
elif tool_shed_repository and tool_id and tool_version:
route_to_images = f"shed_tool_static/{tool_shed_repository.tool_shed}/{tool_shed_repository.owner}/{tool_shed_repository.name}/{tool_id}/{tool_version}"
else:
raise Exception(
"encoded_repository_id or tool_shed_repository and tool_id and tool_version must be provided"
)
# We used to require $PATH_TO_IMAGES and ${static_path}, but
# we now eliminate it if it's used.
text = text.replace("$PATH_TO_IMAGES", "")
text = text.replace("${static_path}", "")
# Use regex to instantiate routes into the defined image paths, but replace
# paths that start with neither http:// nor https://, which will allow for
# settings like .. images:: http_files/images/help.png
for match in re.findall(".. image:: (?!http)/?(.+)", text):
text = text.replace(match, match.replace("/", "%2F"))
text = re.sub(r"\.\. image:: (?!https?://)/?(.+)", rf".. image:: {route_to_images}/\1", text)
return text
__all__ = (
"can_eliminate_repository_dependency",
"clean_dependency_relationships",
"DATATYPES_CONFIG_FILENAME",
"generate_tool_guid",
"get_ctx_rev",
"get_next_prior_import_or_install_required_dict_entry",
"get_tool_panel_config_tool_path_install_dir",
"get_user",
"have_shed_tool_conf_for_install",
"set_image_paths",
)