Merge pull request #551 from Remi-Gau/simple_type

[MAINT] Simplify type annotations

.github/workflows/pull-request.yml

@@ -1,5 +1,9 @@
 name: CI
 
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
 on:
   pull_request:
     branches: [ master ]
@@ -10,6 +14,7 @@ jobs:
     strategy:
       matrix:
         python-version: ['3.11', '3.10', '3.9', '3.8']
+      fail-fast: false
     steps:
       - name: Install Apptainer
         env:

neurodocker/cli/generate.py

@@ -5,10 +5,12 @@
 
 # TODO: add a dedicated class for key=value in the eat-all class.
 
+from __future__ import annotations
+
 import json as json_lib
 import sys
-import typing as ty
 from pathlib import Path
+from typing import IO, Any, Optional, Type, cast
 
 import click
 
@@ -48,15 +50,15 @@ def __init__(self, *args, **kwds):
             )
         ]
 
-    def get_command(self, ctx: click.Context, name: str) -> ty.Optional[click.Command]:
+    def get_command(self, ctx: click.Context, name: str) -> Optional[click.Command]:
         command = self.commands.get(name)
         if command is None:
             return command  # return immediately to error can be logged
 
         # This is only set if a subcommand is called. Calling --help on the group
         # does not set --template-path.
-        template_path: ty.Tuple[str] = ctx.params.get("template_path", tuple())
-        yamls: ty.List[Path] = []
+        template_path: tuple[str] = ctx.params.get("template_path", tuple())
+        yamls: list[Path] = []
         for p in template_path:
             path = Path(p)
             for pattern in ("*.yaml", "*.yml"):
@@ -65,7 +67,7 @@ def get_command(self, ctx: click.Context, name: str) -> ty.Optional[click.Comman
         for path in yamls:
             _ = register_template(path)
 
-        params: ty.List[click.Parameter] = [
+        params: list[click.Parameter] = [
             click.Option(
                 ["-p", "--pkg-manager"],
                 type=click.Choice(list(allowed_pkg_managers), case_sensitive=False),
@@ -86,11 +88,11 @@ class OrderedParamsCommand(click.Command):
     parameters.
     """
 
-    def parse_args(self, ctx: click.Context, args: ty.List[str]):
-        self._options: ty.List[ty.Tuple[click.Parameter, ty.Any]] = []
+    def parse_args(self, ctx: click.Context, args: list[str]):
+        self._options: list[tuple[click.Parameter, Any]] = []
         # run the parser for ourselves to preserve the passed order
         parser = self.make_parser(ctx)
-        param_order: ty.List[click.Parameter]
+        param_order: list[click.Parameter]
         opts, _, param_order = parser.parse_args(args=list(args))
         for param in param_order:
             # We need the parameter name... so if it's None, let's panic.
@@ -186,8 +188,8 @@ def fn(v: str):
             return fn(value)
 
 
-def _get_common_renderer_params() -> ty.List[click.Parameter]:
-    params: ty.List[click.Parameter] = [
+def _get_common_renderer_params() -> list[click.Parameter]:
+    params: list[click.Parameter] = [
         click.Option(
             ["-p", "--pkg-manager"],
             type=click.Choice(list(allowed_pkg_managers), case_sensitive=False),
@@ -290,9 +292,9 @@ def _create_help_for_template(template: Template) -> str:
     return h
 
 
-def _get_params_for_registered_templates() -> ty.List[click.Parameter]:
+def _get_params_for_registered_templates() -> list[click.Parameter]:
     """Return list of click parameters for registered templates."""
-    params: ty.List[click.Parameter] = []
+    params: list[click.Parameter] = []
     names_tmpls = list(registered_templates_items())
     names_tmpls.sort(key=lambda r: r[0])  # sort by name
     for name, tmpl in names_tmpls:
@@ -308,7 +310,7 @@ def _params_to_renderer_dict(ctx: click.Context, pkg_manager) -> dict:
     """Return dictionary compatible with compatible with `_Renderer.from_dict()`."""
     renderer_dict = {"pkg_manager": pkg_manager, "instructions": []}
     cmd = ctx.command
-    cmd = ty.cast(OrderedParamsCommand, cmd)
+    cmd = cast(OrderedParamsCommand, cmd)
     for param, value in cmd._options:
         d = _get_instruction_for_param(ctx=ctx, param=param, value=value)
         # TODO: what happens if `d is None`?
@@ -319,9 +321,7 @@ def _params_to_renderer_dict(ctx: click.Context, pkg_manager) -> dict:
     return renderer_dict
 
 
-def _get_instruction_for_param(
-    ctx: click.Context, param: click.Parameter, value: ty.Any
-):
+def _get_instruction_for_param(ctx: click.Context, param: click.Parameter, value: Any):
     # TODO: clean this up.
     d = None
     if param.name == "from_":
@@ -415,7 +415,7 @@ def generate(*, template_path):
 
 
 def _base_generate(
-    ctx: click.Context, renderer: ty.Type[_Renderer], pkg_manager: str, **kwds
+    ctx: click.Context, renderer: Type[_Renderer], pkg_manager: str, **kwds
 ):
     """Function that does all of the work of `generate docker` and
     `generate singularity`. The difference between those two is the renderer used.
@@ -475,14 +475,14 @@ def singularity(ctx: click.Context, pkg_manager: str, **kwds):
     type=click.File("r"),
     default=sys.stdin,
 )
-def genfromjson(*, container_type: str, input: ty.IO):
+def genfromjson(*, container_type: str, input: IO):
     """Generate a container from a ReproEnv JSON file.
 
     INPUT is standard input by default or a path to a JSON file.
     """
     d = json_lib.load(input)
 
-    renderer: ty.Type[_Renderer]
+    renderer: Type[_Renderer]
     if container_type.lower() == "docker":
         renderer = DockerRenderer
     elif container_type.lower() == "singularity":

neurodocker/cli/minify/_prune.py

@@ -1,6 +1,6 @@
 """Remove all files under a directory but not caught by `reprozip trace`."""
+from __future__ import annotations
 
-import typing as ty
 from pathlib import Path
 
 import yaml
@@ -18,8 +18,8 @@ def _in_docker() -> bool:
 
 def main(
     *,
-    yaml_file: ty.Union[str, Path],
-    directories_to_prune: ty.Union[ty.List[str], ty.List[Path]],
+    yaml_file: str | Path,
+    directories_to_prune: list[str] | list[Path],
 ):
     if not _in_docker():
         raise RuntimeError(
@@ -50,7 +50,7 @@ def main(
         if not d.is_dir():
             raise ValueError(f"Directory does not exist: {d}")
 
-    all_files: ty.Set[Path] = set()
+    all_files: set[Path] = set()
     for d in directories_to_prune:
         all_files.update(Path(d).rglob("*"))
 

neurodocker/cli/minify/trace.py

@@ -7,11 +7,13 @@
 # TODO: consider implementing custom types for Docker container and paths within a
 # Docker container.
 
+from __future__ import annotations
+
 import io
 import logging
 import tarfile
-import typing as ty
 from pathlib import Path
+from typing import Generator, cast
 
 import click
 
@@ -36,9 +38,9 @@
 
 
 def copy_file_to_container(
-    container: ty.Union[str, docker.models.containers.Container],
-    src: ty.Union[str, Path],
-    dest: ty.Union[str, Path],
+    container: str | docker.models.containers.Container,
+    src: str | Path,
+    dest: str | Path,
 ) -> bool:
     """Copy `local_filepath` into `container`:`container_path`.
 
@@ -98,10 +100,10 @@ def _get_mounts(container: docker.models.containers.Container) -> dict:
 @click.option("--yes", is_flag=True, help="Reply yes to all prompts.")
 @click.argument("command", nargs=-1, required=True)
 def minify(
-    container: ty.Union[str, docker.models.containers.Container],
-    directories_to_prune: ty.Tuple[str],
+    container: str | docker.models.containers.Container,
+    directories_to_prune: tuple[str],
     yes: bool,
-    command: ty.Tuple[str],
+    command: tuple[str],
 ) -> None:
     """Minify a container.
 
@@ -118,7 +120,7 @@ def minify(
         "python -c 'a = 1 + 1; print(a)'"
     """
     container = client.containers.get(container)
-    container = ty.cast(docker.models.containers.Container, container)
+    container = cast(docker.models.containers.Container, container)
 
     cmds = " ".join(f'"{c}"' for c in command)
 
@@ -132,7 +134,7 @@ def minify(
     # iteration.
     exec_dict: dict = container.client.api.exec_create(container.id, cmd=trace_cmd)
     exec_id: str = exec_dict["Id"]
-    log_gen: ty.Generator[bytes, None, None] = container.client.api.exec_start(
+    log_gen: Generator[bytes, None, None] = container.client.api.exec_start(
         exec_id, stream=True
     )
     for log in log_gen: