Merge pull request #2223 from NovakApis/feature/python-typing/compone…

…nts_command.py

Feature: Implement typing for files components_command.py and components_utils.py. Set static type checking scope to changed files only.

.github/workflows/lint-code.yml

@@ -85,3 +85,22 @@ jobs:
         with:
           isortVersion: "latest"
           requirementsFiles: "requirements.txt requirements-dev.txt"
+
+  static-type-check:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - uses: actions/setup-python@v3
+        with:
+          python-version: "3.11"
+      - run: pip install mypy types-PyYAML
+      - name: Get Python changed files
+        id: changed-py-files
+        uses: tj-actions/changed-files@v23
+        with:
+          files: |
+            *.py
+            **/*.py
+      - name: Run if any of the listed files above is changed
+        if: steps.changed-py-files.outputs.any_changed == 'true'
+        run: mypy ${{ steps.changed-py-files.outputs.all_changed_files }} --ignore-missing-imports

nf_core/components/components_command.py

@@ -3,8 +3,7 @@
 import os
 import shutil
 from pathlib import Path
-
-import yaml
+from typing import Dict, List, Optional, Union
 
 import nf_core.utils
 from nf_core.modules.modules_json import ModulesJson
@@ -20,7 +19,15 @@ class ComponentCommand:
     Base class for the 'nf-core modules' and 'nf-core subworkflows' commands
     """
 
-    def __init__(self, component_type, dir, remote_url=None, branch=None, no_pull=False, hide_progress=False):
+    def __init__(
+        self,
+        component_type: str,
+        dir: str,
+        remote_url: Optional[str] = None,
+        branch: Optional[str] = None,
+        no_pull: bool = False,
+        hide_progress: bool = False,
+    ) -> None:
         """
         Initialise the ComponentClass object
         """
@@ -30,14 +37,15 @@ def __init__(self, component_type, dir, remote_url=None, branch=None, no_pull=Fa
         self.hide_progress = hide_progress
         self._configure_repo_and_paths()
 
-    def _configure_repo_and_paths(self, nf_dir_req=True):
+    def _configure_repo_and_paths(self, nf_dir_req: bool = True) -> None:
         """
         Determine the repo type and set some default paths.
         If this is a modules repo, determine the org_path too.
 
         Args:
             nf_dir_req (bool, optional): Whether this command requires being run in the nf-core modules repo or a nf-core pipeline repository. Defaults to True.
         """
+
         try:
             if self.dir:
                 self.dir, self.repo_type, self.org = get_repo_info(self.dir, use_prompt=nf_dir_req)
@@ -54,7 +62,7 @@ def _configure_repo_and_paths(self, nf_dir_req=True):
         self.default_subworkflows_path = Path("subworkflows", self.org)
         self.default_subworkflows_tests_path = Path("tests", "subworkflows", self.org)
 
-    def get_local_components(self):
+    def get_local_components(self) -> List[str]:
         """
         Get the local modules/subworkflows in a pipeline
         """
@@ -63,7 +71,7 @@ def get_local_components(self):
             str(path.relative_to(local_component_dir)) for path in local_component_dir.iterdir() if path.suffix == ".nf"
         ]
 
-    def get_components_clone_modules(self):
+    def get_components_clone_modules(self) -> List[str]:
         """
         Get the modules/subworkflows repository available in a clone of nf-core/modules
         """
@@ -77,7 +85,7 @@ def get_components_clone_modules(self):
             if "main.nf" in files
         ]
 
-    def has_valid_directory(self):
+    def has_valid_directory(self) -> bool:
         """Check that we were given a pipeline or clone of nf-core/modules"""
         if self.repo_type == "modules":
             return True
@@ -92,14 +100,14 @@ def has_valid_directory(self):
             log.warning(f"Could not find a 'main.nf' or 'nextflow.config' file in '{self.dir}'")
         return True
 
-    def has_modules_file(self):
+    def has_modules_file(self) -> None:
         """Checks whether a module.json file has been created and creates one if it is missing"""
         modules_json_path = os.path.join(self.dir, "modules.json")
         if not os.path.exists(modules_json_path):
             log.info("Creating missing 'module.json' file.")
             ModulesJson(self.dir).create()
 
-    def clear_component_dir(self, component_name, component_dir):
+    def clear_component_dir(self, component_name: str, component_dir: str) -> bool:
         """
         Removes all files in the module/subworkflow directory
 
@@ -127,7 +135,7 @@ def clear_component_dir(self, component_name, component_dir):
             log.error(f"Could not remove {self.component_type[:-1]} {component_name}: {e}")
             return False
 
-    def components_from_repo(self, install_dir):
+    def components_from_repo(self, install_dir: str) -> List[str]:
         """
         Gets the modules/subworkflows installed from a certain repository
 
@@ -145,7 +153,9 @@ def components_from_repo(self, install_dir):
             str(Path(dir_path).relative_to(repo_dir)) for dir_path, _, files in os.walk(repo_dir) if "main.nf" in files
         ]
 
-    def install_component_files(self, component_name, component_version, modules_repo, install_dir):
+    def install_component_files(
+        self, component_name: str, component_version: str, modules_repo: ModulesRepo, install_dir: str
+    ) -> bool:
         """
         Installs a module/subworkflow into the given directory
 
@@ -160,7 +170,7 @@ def install_component_files(self, component_name, component_version, modules_rep
         """
         return modules_repo.install_component(component_name, install_dir, component_version, self.component_type)
 
-    def load_lint_config(self):
+    def load_lint_config(self) -> None:
         """Parse a pipeline lint config file.
 
         Load the '.nf-core.yml'  config file and extract
@@ -171,7 +181,7 @@ def load_lint_config(self):
         _, tools_config = nf_core.utils.load_tools_config(self.dir)
         self.lint_config = tools_config.get("lint", {})
 
-    def check_modules_structure(self):
+    def check_modules_structure(self) -> None:
         """
         Check that the structure of the modules directory in a pipeline is the correct one:
             modules/nf-core/TOOL/SUBTOOL
@@ -180,7 +190,7 @@ def check_modules_structure(self):
             modules/nf-core/modules/TOOL/SUBTOOL
         """
         if self.repo_type == "pipeline":
-            wrong_location_modules = []
+            wrong_location_modules: List[Path] = []
             for directory, _, files in os.walk(Path(self.dir, "modules")):
                 if "main.nf" in files:
                     module_path = Path(directory).relative_to(Path(self.dir, "modules"))
@@ -201,14 +211,14 @@ def check_modules_structure(self):
                     modules_dir = Path("modules").resolve()
                     correct_dir = Path(modules_dir, self.modules_repo.repo_path, Path(*module.parts[2:]))
                     wrong_dir = Path(modules_dir, module)
-                    shutil.move(wrong_dir, correct_dir)
+                    shutil.move(str(wrong_dir), str(correct_dir))
                     log.info(f"Moved {wrong_dir} to {correct_dir}.")
                 shutil.rmtree(Path(self.dir, "modules", self.modules_repo.repo_path, "modules"))
                 # Regenerate modules.json file
                 modules_json = ModulesJson(self.dir)
                 modules_json.check_up_to_date()
 
-    def check_patch_paths(self, patch_path, module_name):
+    def check_patch_paths(self, patch_path: Path, module_name: str) -> None:
         """
         Check that paths in patch files are updated to the new modules path
         """
@@ -239,7 +249,7 @@ def check_patch_paths(self, patch_path, module_name):
                     ][module_name]["patch"] = str(patch_path.relative_to(Path(self.dir).resolve()))
                 modules_json.dump()
 
-    def check_if_in_include_stmts(self, component_path):
+    def check_if_in_include_stmts(self, component_path: str) -> Dict[str, List[Dict[str, Union[int, str]]]]:
         """
         Checks for include statements in the main.nf file of the pipeline and a list of line numbers where the component is included
         Args:
@@ -248,7 +258,7 @@ def check_if_in_include_stmts(self, component_path):
         Returns:
             (list): A list of dictionaries, with the workflow file and the line number where the component is included
         """
-        include_stmts = {}
+        include_stmts: Dict[str, List[Dict[str, Union[int, str]]]] = {}
         if self.repo_type == "pipeline":
             workflow_files = Path(self.dir, "workflows").glob("*.nf")
             for workflow_file in workflow_files:

nf_core/components/components_utils.py

@@ -1,31 +1,33 @@
 import logging
-import os
 import re
 from pathlib import Path
+from typing import List, Optional, Tuple
 
 import questionary
 import rich.prompt
 
 import nf_core.utils
+from nf_core.modules.modules_repo import ModulesRepo
 
 log = logging.getLogger(__name__)
 
 
-def get_repo_info(directory, use_prompt=True):
+def get_repo_info(directory: str, use_prompt: Optional[bool] = True) -> Tuple[str, Optional[str], str]:
     """
     Determine whether this is a pipeline repository or a clone of
     nf-core/modules
     """
+
     # Verify that the pipeline dir exists
     if directory is None or not Path(directory).is_dir():
         raise UserWarning(f"Could not find directory: {directory}")
 
     # Try to find the root directory
-    base_dir = nf_core.utils.determine_base_dir(directory)
+    base_dir: str = nf_core.utils.determine_base_dir(directory)
 
     # Figure out the repository type from the .nf-core.yml config file if we can
     config_fn, tools_config = nf_core.utils.load_tools_config(base_dir)
-    repo_type = tools_config.get("repository_type", None)
+    repo_type: Optional[str] = tools_config.get("repository_type", None)
 
     # If not set, prompt the user
     if not repo_type and use_prompt:
@@ -55,7 +57,6 @@ def get_repo_info(directory, use_prompt=True):
         raise UserWarning(f"Invalid repository type: '{repo_type}'")
 
     # Check for org if modules repo
-    org = None
     if repo_type == "pipeline":
         org = ""
     elif repo_type == "modules":
@@ -77,10 +78,12 @@ def get_repo_info(directory, use_prompt=True):
             raise UserWarning("Organisation path could not be established")
 
     # It was set on the command line, return what we were given
-    return [base_dir, repo_type, org]
+    return (base_dir, repo_type, org)
 
 
-def prompt_component_version_sha(component_name, component_type, modules_repo, installed_sha=None):
+def prompt_component_version_sha(
+    component_name: str, component_type: str, modules_repo: ModulesRepo, installed_sha: Optional[str] = None
+) -> str:
     """
     Creates an interactive questionary prompt for selecting the module/subworkflow version
     Args:
@@ -107,17 +110,20 @@ def prompt_component_version_sha(component_name, component_type, modules_repo, i
         next_page_commits = [next(all_commits, None) for _ in range(10)]
         next_page_commits = [commit for commit in next_page_commits if commit is not None]
         if all(commit is None for commit in next_page_commits):
-            next_page_commits = None
+            next_page_commits = []
 
         choices = []
-        for title, sha in map(lambda commit: (commit["trunc_message"], commit["git_sha"]), commits):
-            display_color = "fg:ansiblue" if sha != installed_sha else "fg:ansired"
-            message = f"{title} {sha}"
-            if installed_sha == sha:
-                message += " (installed version)"
-            commit_display = [(display_color, message), ("class:choice-default", "")]
-            choices.append(questionary.Choice(title=commit_display, value=sha))
-        if next_page_commits is not None:
+        for commit in commits:
+            if commit:
+                title = commit["trunc_message"]
+                sha = commit["git_sha"]
+                display_color = "fg:ansiblue" if sha != installed_sha else "fg:ansired"
+                message = f"{title} {sha}"
+                if installed_sha == sha:
+                    message += " (installed version)"
+                commit_display = [(display_color, message), ("class:choice-default", "")]
+                choices.append(questionary.Choice(title=commit_display, value=sha))
+        if next_page_commits:
             choices += [older_commits_choice]
         git_sha = questionary.select(
             f"Select '{component_name}' commit:", choices=choices, style=nf_core.utils.nfcore_question_style
@@ -126,7 +132,7 @@ def prompt_component_version_sha(component_name, component_type, modules_repo, i
     return git_sha
 
 
-def get_components_to_install(subworkflow_dir):
+def get_components_to_install(subworkflow_dir: str) -> Tuple[List[str], List[str]]:
     """
     Parse the subworkflow main.nf file to retrieve all imported modules and subworkflows.
     """

pyproject.toml

@@ -20,3 +20,8 @@ norecursedirs = [ ".*", "build", "dist", "*.egg", "data", "__pycache__", ".githu
 profile = "black"
 known_first_party = ["nf_core"]
 multi_line_output = 3
+
+[tool.mypy]
+ignore_missing_imports = true
+follow_imports = "skip"
+disable_error_code = "no-redef"