diff --git a/planemo/commands/cmd_container_register.py b/planemo/commands/cmd_container_register.py index f4330b499..685b0ceb5 100644 --- a/planemo/commands/cmd_container_register.py +++ b/planemo/commands/cmd_container_register.py @@ -1,6 +1,7 @@ """Module describing the planemo ``container_register`` command.""" import os import string +from typing import List import click from galaxy.tool_util.deps.container_resolvers.mulled import targets_to_mulled_name @@ -10,11 +11,15 @@ ) from galaxy.tool_util.deps.mulled.util import ( conda_build_target_str, + CondaTarget, v2_image_name, ) from planemo import options -from planemo.cli import command_function +from planemo.cli import ( + command_function, + PlanemoCliContext, +) from planemo.conda import ( best_practice_search, build_conda_context, @@ -78,7 +83,7 @@ help="Force push branch for pull request in case it already exists.", ) @command_function -def cli(ctx, paths, **kwds): +def cli(ctx: "PlanemoCliContext", paths, **kwds) -> None: """Register multi-requirement containers as needed. BioContainers publishes all Bioconda packages automatically as individual @@ -154,7 +159,7 @@ def cli(ctx, paths, **kwds): class RegistryTarget: """Abstraction around mulled container registry (both directory and Github repo).""" - def __init__(self, ctx, **kwds): + def __init__(self, ctx: "PlanemoCliContext", **kwds): output_directory = kwds["output_directory"] pr_titles = [] target_repository = None @@ -179,7 +184,7 @@ def __init__(self, ctx, **kwds): self.output_directory = output_directory self.target_repository = target_repository - def has_pull_request_for(self, name): + def has_pull_request_for(self, name: str) -> bool: has_pr = False if self.do_pull_request: if any([name in t for t in self.pr_titles]): @@ -187,7 +192,16 @@ def has_pull_request_for(self, name): return has_pr - def handle_pull_request(self, ctx, name, target_filename, packages_str, tools_str, base_image, **kwds): + def handle_pull_request( + self, + ctx: "PlanemoCliContext", + name: str, + target_filename: str, + packages_str: str, + tools_str: str, + base_image: str, + **kwds, + ) -> None: if self.do_pull_request: message = kwds["message"] message = string.Template(message).safe_substitute( @@ -199,6 +213,7 @@ def handle_pull_request(self, ctx, name, target_filename, packages_str, tools_st } ) branch_name = name.replace(":", "-") + assert self.target_repository branch(ctx, self.target_repository, branch_name, from_branch="master") add(ctx, self.target_repository, target_filename) commit(ctx, self.target_repository, message=message) @@ -206,25 +221,32 @@ def handle_pull_request(self, ctx, name, target_filename, packages_str, tools_st push(ctx, repo_path=self.target_repository, to=self.remote_name, branch=branch_name, force=force_push) pull_request(ctx, self.target_repository, message=message, repo=REGISTRY_REPOSITORY) - def write_targets(self, ctx, target_filename, mulled_targets, tag, base_image): + def write_targets( + self, + ctx: "PlanemoCliContext", + target_filename: str, + mulled_targets: List[CondaTarget], + tag: str, + base_image: str, + ) -> None: with open(target_filename, "w") as f: targets = to_target_str(mulled_targets) f.write(string.Template(CONTENTS).safe_substitute(targets=targets, base_image=base_image, image_build=tag)) ctx.log(f"Wrote requirements [{targets}] to file [{target_filename}]") -def to_target_str(targets): +def to_target_str(targets: List[CondaTarget]) -> str: target_strings = [] for target in targets: if target.version: - target_str = f"{target.package_name}={target.version}" + target_str = f"{target.package}={target.version}" else: - target_str = target.package_name + target_str = target.package target_strings.append(target_str) return ",".join(target_strings) -def open_prs(ctx): +def open_prs(ctx: "PlanemoCliContext") -> List: repo = get_repository_object(ctx, REGISTRY_REPOSITORY) prs = [pr for pr in repo.get_pulls()] return prs diff --git a/planemo/conda.py b/planemo/conda.py index 79fedf65b..a338b2025 100644 --- a/planemo/conda.py +++ b/planemo/conda.py @@ -8,10 +8,24 @@ import os import threading from copy import deepcopy -from typing import TYPE_CHECKING +from typing import ( + Any, + Dict, + FrozenSet, + Iterable, + List, + Optional, + Set, + Tuple, + TYPE_CHECKING, + Union, +) from galaxy.tool_util.deps import conda_util -from galaxy.tool_util.deps.conda_util import CondaContext +from galaxy.tool_util.deps.conda_util import ( + CondaContext, + CondaTarget, +) from galaxy.util import unicodify from planemo.exit_codes import ( @@ -43,7 +57,7 @@ def build_conda_context(ctx: "PlanemoCliContext", **kwds) -> CondaContext: condarc_override = kwds.get("condarc", condarc_override_default) use_local = kwds.get("conda_use_local", False) shell_exec = shell if use_planemo_shell else None - conda_context = conda_util.CondaContext( + conda_context = CondaContext( conda_prefix=conda_prefix, ensure_channels=ensure_channels, condarc_override=condarc_override, @@ -100,21 +114,23 @@ def collect_conda_targets(ctx, paths, recursive=False, found_tool_callback=None) # Copied and modified from mulled stuff - need to syncronize these concepts. -def target_str_to_targets(targets_raw): - def parse_target(target_str): +def target_str_to_targets(targets_raw: str) -> List[CondaTarget]: + def parse_target(target_str: str) -> CondaTarget: if "=" in target_str: package_name, version = target_str.split("=", 1) else: package_name = target_str version = None - target = conda_util.CondaTarget(package_name, version) + target = CondaTarget(package_name, version) return target targets = [parse_target(_) for _ in targets_raw.split(",")] return targets -def collect_conda_target_lists(ctx, paths, recursive=False, found_tool_callback=None): +def collect_conda_target_lists( + ctx: "PlanemoCliContext", paths: Iterable[str], recursive: bool = False, found_tool_callback=None +) -> List[FrozenSet[CondaTarget]]: """Load CondaTarget lists from supplied artifact sources. If a tool contains more than one requirement, the requirements will all @@ -126,26 +142,28 @@ def collect_conda_target_lists(ctx, paths, recursive=False, found_tool_callback= return conda_target_lists -def collect_conda_target_lists_and_tool_paths(ctx, paths, recursive=False, found_tool_callback=None): +def collect_conda_target_lists_and_tool_paths( + ctx: "PlanemoCliContext", paths: Iterable[str], recursive: bool = False, found_tool_callback=None +) -> Tuple[List[FrozenSet[CondaTarget]], List[List[str]]]: """Load CondaTarget lists from supplied artifact sources. If a tool contains more than one requirement, the requirements will all appear together as one list element of the output list. """ - conda_target_lists = set() + conda_target_sets: Set[FrozenSet[CondaTarget]] = set() tool_paths = collections.defaultdict(list) for tool_path, tool_source in yield_tool_sources_on_paths(ctx, paths, recursive=recursive, yield_load_errors=False): try: if found_tool_callback: found_tool_callback(tool_path) targets = frozenset(tool_source_conda_targets(tool_source)) - conda_target_lists.add(targets) + conda_target_sets.add(targets) tool_paths[targets].append(tool_path) except Exception as e: ctx.log(f"Error while collecting list of conda targets for '{tool_path}': {unicodify(e)}") # Turn them into lists so the order matches before returning... - conda_target_lists = list(conda_target_lists) + conda_target_lists = list(conda_target_sets) conda_target_tool_paths = [tool_paths[c] for c in conda_target_lists] return conda_target_lists, conda_target_tool_paths @@ -160,7 +178,9 @@ def tool_source_conda_targets(tool_source): best_practice_search_first = threading.local() -def best_practice_search(conda_target, conda_context=None, platform=None): +def best_practice_search( + conda_target: CondaTarget, conda_context: Optional[CondaContext] = None, platform: Optional[str] = None +) -> Union[Tuple[None, None], Tuple[Dict[str, Any], bool]]: # Call it in offline mode after the first time. try: best_practice_search_first.previously_called @@ -175,7 +195,7 @@ def best_practice_search(conda_target, conda_context=None, platform=None): conda_context = deepcopy(conda_context) conda_context.ensure_channels = BEST_PRACTICE_CHANNELS else: - conda_context = conda_util.CondaContext(ensure_channels=BEST_PRACTICE_CHANNELS) + conda_context = CondaContext(ensure_channels=BEST_PRACTICE_CHANNELS) return conda_util.best_search_result( conda_target, conda_context=conda_context, diff --git a/planemo/mulled.py b/planemo/mulled.py index 510fd15be..e3f588eb6 100644 --- a/planemo/mulled.py +++ b/planemo/mulled.py @@ -4,23 +4,30 @@ """ import os +from typing import ( + Iterable, + List, +) from galaxy.tool_util.deps.mulled.mulled_build import ( DEFAULT_CHANNELS, ensure_installed, InvolucroContext, ) -from galaxy.tool_util.deps.mulled.util import build_target +from galaxy.tool_util.deps.mulled.util import ( + build_target, + CondaTarget, +) from planemo.conda import collect_conda_target_lists from planemo.io import shell -def conda_to_mulled_targets(conda_targets): +def conda_to_mulled_targets(conda_targets: Iterable[CondaTarget]) -> List[CondaTarget]: return list(map(lambda c: build_target(c.package, c.version), conda_targets)) -def collect_mulled_target_lists(ctx, paths, recursive=False): +def collect_mulled_target_lists(ctx, paths: Iterable[str], recursive: bool = False) -> List[List[CondaTarget]]: return list(map(conda_to_mulled_targets, collect_conda_target_lists(ctx, paths, recursive=recursive))) diff --git a/requirements.txt b/requirements.txt index 080f8f932..e32d13620 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,8 +3,8 @@ bioblend>=1.0.0 click!=8.0.2 cwltool>=1.0.20191225192155 ephemeris>=0.10.3 -galaxy-tool-util>=23.0,<24.0 -galaxy-util[template]>=23.0,<24.0 +galaxy-tool-util>=23.1,<24.0 +galaxy-util[template]>=23.1,<24.0 glob2 gxformat2>=0.14.0 h5py @@ -15,4 +15,4 @@ pathvalidate pyyaml virtualenv stdlib_list; python_version < '3.10' -tabulate \ No newline at end of file +tabulate