import os
from typing import (
Any,
Dict,
)
import requests
from planemo.galaxy.profiles import translate_alias
from planemo.galaxy.workflows import (
GALAXY_WORKFLOW_INSTANCE_PREFIX,
GALAXY_WORKFLOWS_PREFIX,
)
from planemo.tools import uri_to_path
from .runnable import (
for_path,
for_uri,
GALAXY_TOOLS_PREFIX,
TRS_WORKFLOWS_PREFIX,
)
[docs]
def for_runnable_identifier(ctx, runnable_identifier, kwds: Dict[str, Any]):
"""Convert URI, path, or alias into Runnable."""
# could be a URI, path, or alias
current_profile = kwds.get("profile")
runnable_identifier = translate_alias(ctx, runnable_identifier, current_profile)
# Check if it's a full Dockstore TRS URL
if runnable_identifier.startswith("https://dockstore.org/api/ga4gh/trs/v2/tools/"):
# Extract the TRS tool ID from the URL
# Format: https://dockstore.org/api/ga4gh/trs/v2/tools/#workflow/github.com/org/repo/workflow_name[/versions/version]
trs_base = "https://dockstore.org/api/ga4gh/trs/v2/tools/"
trs_path = runnable_identifier[len(trs_base) :]
# Convert to TRS URI by wrapping with trs://
runnable_identifier = f"{TRS_WORKFLOWS_PREFIX}{trs_path}"
return for_uri(runnable_identifier)
# Check if it's a TRS ID - convert to TRS URI (don't download)
# Support both formats: workflow/... and #workflow/...
is_trs_id = (
runnable_identifier.startswith(("workflow/", "tool/", "#workflow/", "#tool/"))
and "/github.com/" in runnable_identifier
)
if is_trs_id:
# This is a TRS ID, convert to TRS URI
runnable_identifier = f"{TRS_WORKFLOWS_PREFIX}{runnable_identifier}"
return for_uri(runnable_identifier)
if not runnable_identifier.startswith(
(GALAXY_WORKFLOWS_PREFIX, GALAXY_WORKFLOW_INSTANCE_PREFIX, TRS_WORKFLOWS_PREFIX)
):
runnable_identifier = uri_to_path(ctx, runnable_identifier)
if os.path.exists(runnable_identifier):
runnable = for_path(runnable_identifier)
else: # assume galaxy workflow or tool id
if "/repos/" in runnable_identifier:
runnable_identifier = f"{GALAXY_TOOLS_PREFIX}{runnable_identifier}"
elif not runnable_identifier.startswith("gxid://") and not runnable_identifier.startswith("trs://"):
runnable_identifier = f"{GALAXY_WORKFLOWS_PREFIX}{runnable_identifier}"
runnable = for_uri(runnable_identifier)
return runnable
[docs]
def for_runnable_identifiers(ctx, runnable_identifiers, kwds: Dict[str, Any]):
"""Convert lists of URIs, paths, and/or aliases into Runnables."""
runnables = []
for r in runnable_identifiers:
runnable = for_runnable_identifier(ctx, r, kwds)
if isinstance(runnable, list):
runnables.extend(runnable)
else:
runnables.append(runnable)
return runnables
[docs]
def install_args_list_to_runnables(ctx, install_args_list, kwds):
runnables = []
for repo in install_args_list:
base_tool_shed_url = repo["tool_shed_url"].rstrip("/")
url = f"{base_tool_shed_url}/api/repositories/get_repository_revision_install_info"
response = requests.get(
url, params={"name": repo["name"], "owner": repo["owner"], "changeset_revision": repo["changeset_revision"]}
)
response.raise_for_status()
install_info = response.json()
repository_metadata = install_info[1]
assert repository_metadata and repository_metadata["model_class"] == "RepositoryMetadata", repository_metadata
for tool in repository_metadata.get("valid_tools", []):
runnable = for_runnable_identifier(ctx, tool["guid"], kwds)
runnables.append(runnable)
return runnables