Fix use of package_name attribute on CondaTarget objects

In Galaxy 23.1 `galaxy.tool_util.deps.mulled.util.build_target()`
was changed to return a `CondaTarget` object instead of a (removed)
`Target`.

To avoid complications, this requires a bump to the galaxy-util and
galaxy-tool-util packages to >=23.1 .

Also, add type annotations to most related code.

Fix galaxyproject/galaxy#17378 .

Author: nsoranzo

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,15 +184,24 @@ 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]):
                 has_pr = True
 
         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,32 +213,40 @@ 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)
             force_push = kwds.get("force_push", False)
             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

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,

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)))
 
 

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
+tabulate