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.util.search_util
import logging
from sqlalchemy import (
false,
select,
true,
)
log = logging.getLogger(__name__)
[docs]def in_tool_dict(tool_dict, exact_matches_checked, tool_id=None, tool_name=None, tool_version=None):
found = False
if tool_id and not tool_name and not tool_version:
tool_dict_tool_id = tool_dict["id"].lower()
found = (tool_id == tool_dict_tool_id) or (not exact_matches_checked and tool_dict_tool_id.find(tool_id) >= 0)
elif tool_name and not tool_id and not tool_version:
tool_dict_tool_name = tool_dict["name"].lower()
found = (tool_name == tool_dict_tool_name) or (
not exact_matches_checked and tool_dict_tool_name.find(tool_name) >= 0
)
elif tool_version and not tool_id and not tool_name:
tool_dict_tool_version = tool_dict["version"].lower()
found = (tool_version == tool_dict_tool_version) or (
not exact_matches_checked and tool_dict_tool_version.find(tool_version) >= 0
)
elif tool_id and tool_name and not tool_version:
tool_dict_tool_id = tool_dict["id"].lower()
tool_dict_tool_name = tool_dict["name"].lower()
found = (tool_id == tool_dict_tool_id and tool_name == tool_dict_tool_name) or (
not exact_matches_checked
and tool_dict_tool_id.find(tool_id) >= 0
and tool_dict_tool_name.find(tool_name) >= 0
)
elif tool_id and tool_version and not tool_name:
tool_dict_tool_id = tool_dict["id"].lower()
tool_dict_tool_version = tool_dict["version"].lower()
found = (tool_id == tool_dict_tool_id and tool_version == tool_dict_tool_version) or (
not exact_matches_checked
and tool_dict_tool_id.find(tool_id) >= 0
and tool_dict_tool_version.find(tool_version) >= 0
)
elif tool_version and tool_name and not tool_id:
tool_dict_tool_version = tool_dict["version"].lower()
tool_dict_tool_name = tool_dict["name"].lower()
found = (tool_version == tool_dict_tool_version and tool_name == tool_dict_tool_name) or (
not exact_matches_checked
and tool_dict_tool_version.find(tool_version) >= 0
and tool_dict_tool_name.find(tool_name) >= 0
)
elif tool_version and tool_name and tool_id:
tool_dict_tool_version = tool_dict["version"].lower()
tool_dict_tool_name = tool_dict["name"].lower()
tool_dict_tool_id = tool_dict["id"].lower()
found = (
tool_version == tool_dict_tool_version and tool_name == tool_dict_tool_name and tool_id == tool_dict_tool_id
) or (
not exact_matches_checked
and tool_dict_tool_version.find(tool_version) >= 0
and tool_dict_tool_name.find(tool_name) >= 0
and tool_dict_tool_id.find(tool_id) >= 0
)
return found
[docs]def make_same_length(list1, list2):
# If either list is 1 item, we'll append to it until its length is the same as the other.
if len(list1) == 1:
for _ in range(1, len(list2)):
list1.append(list1[0])
elif len(list2) == 1:
for _ in range(1, len(list1)):
list2.append(list2[0])
return list1, list2
[docs]def search_ids_names(tool_dict, exact_matches_checked, match_tuples, repository_metadata, tool_ids, tool_names):
for i, tool_id in enumerate(tool_ids):
tool_name = tool_names[i]
if in_tool_dict(tool_dict, exact_matches_checked, tool_id=tool_id, tool_name=tool_name):
match_tuples.append((repository_metadata.repository_id, repository_metadata.changeset_revision))
return match_tuples
[docs]def search_ids_versions(tool_dict, exact_matches_checked, match_tuples, repository_metadata, tool_ids, tool_versions):
for i, tool_id in enumerate(tool_ids):
tool_version = tool_versions[i]
if in_tool_dict(tool_dict, exact_matches_checked, tool_id=tool_id, tool_version=tool_version):
match_tuples.append((repository_metadata.repository_id, repository_metadata.changeset_revision))
return match_tuples
[docs]def search_names_versions(
tool_dict, exact_matches_checked, match_tuples, repository_metadata, tool_names, tool_versions
):
for i, tool_name in enumerate(tool_names):
tool_version = tool_versions[i]
if in_tool_dict(tool_dict, exact_matches_checked, tool_name=tool_name, tool_version=tool_version):
match_tuples.append((repository_metadata.repository_id, repository_metadata.changeset_revision))
return match_tuples
[docs]def search_repository_metadata(app, exact_matches_checked, tool_ids="", tool_names="", tool_versions=""):
sa_session = app.model.session
match_tuples = []
ok = True
if tool_ids or tool_names or tool_versions:
for repository_metadata in get_metadata(sa_session, app.model.RepositoryMetadata, app.model.Repository):
metadata = repository_metadata.metadata
if metadata:
tools = metadata.get("tools", [])
for tool_dict in tools:
if tool_ids and not tool_names and not tool_versions:
for tool_id in tool_ids:
if in_tool_dict(tool_dict, exact_matches_checked, tool_id=tool_id):
match_tuples.append(
(repository_metadata.repository_id, repository_metadata.changeset_revision)
)
elif tool_names and not tool_ids and not tool_versions:
for tool_name in tool_names:
if in_tool_dict(tool_dict, exact_matches_checked, tool_name=tool_name):
match_tuples.append(
(repository_metadata.repository_id, repository_metadata.changeset_revision)
)
elif tool_versions and not tool_ids and not tool_names:
for tool_version in tool_versions:
if in_tool_dict(tool_dict, exact_matches_checked, tool_version=tool_version):
match_tuples.append(
(repository_metadata.repository_id, repository_metadata.changeset_revision)
)
elif tool_ids and tool_names and not tool_versions:
if len(tool_ids) == len(tool_names):
match_tuples = search_ids_names(
tool_dict,
exact_matches_checked,
match_tuples,
repository_metadata,
tool_ids,
tool_names,
)
elif len(tool_ids) == 1 or len(tool_names) == 1:
tool_ids, tool_names = make_same_length(tool_ids, tool_names)
match_tuples = search_ids_names(
tool_dict,
exact_matches_checked,
match_tuples,
repository_metadata,
tool_ids,
tool_names,
)
else:
ok = False
elif tool_ids and tool_versions and not tool_names:
if len(tool_ids) == len(tool_versions):
match_tuples = search_ids_versions(
tool_dict,
exact_matches_checked,
match_tuples,
repository_metadata,
tool_ids,
tool_versions,
)
elif len(tool_ids) == 1 or len(tool_versions) == 1:
tool_ids, tool_versions = make_same_length(tool_ids, tool_versions)
match_tuples = search_ids_versions(
tool_dict,
exact_matches_checked,
match_tuples,
repository_metadata,
tool_ids,
tool_versions,
)
else:
ok = False
elif tool_versions and tool_names and not tool_ids:
if len(tool_versions) == len(tool_names):
match_tuples = search_names_versions(
tool_dict,
exact_matches_checked,
match_tuples,
repository_metadata,
tool_names,
tool_versions,
)
elif len(tool_versions) == 1 or len(tool_names) == 1:
tool_versions, tool_names = make_same_length(tool_versions, tool_names)
match_tuples = search_names_versions(
tool_dict,
exact_matches_checked,
match_tuples,
repository_metadata,
tool_names,
tool_versions,
)
else:
ok = False
elif tool_versions and tool_names and tool_ids:
if len(tool_versions) == len(tool_names) and len(tool_names) == len(tool_ids):
for i, tool_version in enumerate(tool_versions):
tool_name = tool_names[i]
tool_id = tool_ids[i]
if in_tool_dict(
tool_dict,
exact_matches_checked,
tool_id=tool_id,
tool_name=tool_name,
tool_version=tool_version,
):
match_tuples.append(
(repository_metadata.repository_id, repository_metadata.changeset_revision)
)
else:
ok = False
return ok, match_tuples
[docs]def get_metadata(session, repository_metadata_model, repository_model):
stmt = (
select(repository_metadata_model)
.where(repository_metadata_model.includes_tools == true())
.join(repository_model)
.where(repository_model.deleted == false())
.where(repository_model.deprecated == false())
)
return session.scalars(stmt)