Feature/python typing/components command.py

.github/workflows/lint-code.yml

@@ -85,3 +85,23 @@ 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
+    - run: pip install 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

.pre-commit-config.yaml

@@ -11,3 +11,4 @@ repos:
     rev: "v2.7.1"
     hooks:
       - id: prettier
+

nf_core/components/components_command.py

@@ -3,6 +3,7 @@
 import os
 import shutil
 from pathlib import Path
+from typing import Dict, List, Optional, Tuple, Union
 
 import yaml
 
@@ -20,17 +21,25 @@ 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_arg: Optional[Union[str, os.PathLike[str]]],
+        remote_url: Optional[str] = None,
+        branch: Optional[str] = None,
+        no_pull: bool = False,
+        hide_progress: bool = False,
+    ) -> None:
         """
         Initialise the ComponentClass object
         """
         self.component_type = component_type
-        self.dir = dir
-        self.modules_repo = ModulesRepo(remote_url, branch, no_pull, hide_progress)
+        self.dir_arg = dir_arg
+        self.modules_repo: ModulesRepo = ModulesRepo(remote_url, branch, no_pull, hide_progress)
         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: Optional[bool] = True) -> None:
         """
         Determine the repo type and set some default paths.
         If this is a modules repo, determine the org_path too.
@@ -39,8 +48,9 @@ def _configure_repo_and_paths(self, nf_dir_req=True):
             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)
+            if self.dir_arg:
+                self.dir_path: os.PathLike[str]
+                self.dir_path, self.repo_type, self.org = get_repo_info(self.dir_path, use_prompt=nf_dir_req)
             else:
                 self.repo_type = None
                 self.org = ""
@@ -49,57 +59,57 @@ def _configure_repo_and_paths(self, nf_dir_req=True):
                 raise
             self.repo_type = None
             self.org = ""
-        self.default_modules_path = Path("modules", self.org)
-        self.default_tests_path = Path("tests", "modules", self.org)
-        self.default_subworkflows_path = Path("subworkflows", self.org)
-        self.default_subworkflows_tests_path = Path("tests", "subworkflows", self.org)
+        self.default_modules_path: Path = Path("modules", self.org)
+        self.default_tests_path: Path = Path("tests", "modules", self.org)
+        self.default_subworkflows_path: Path = Path("subworkflows", self.org)
+        self.default_subworkflows_tests_path: 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
         """
-        local_component_dir = Path(self.dir, self.component_type, "local")
+        local_component_dir = Path(self.dir_path, self.component_type, "local")
         return [
             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
         """
         if self.component_type == "modules":
-            component_base_path = Path(self.dir, self.default_modules_path)
+            component_base_path = Path(self.dir_path, self.default_modules_path)
         elif self.component_type == "subworkflows":
-            component_base_path = Path(self.dir, self.default_subworkflows_path)
+            component_base_path = Path(self.dir_path, self.default_subworkflows_path)
         return [
             str(Path(dir).relative_to(component_base_path))
             for dir, _, files in os.walk(component_base_path)
             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
-        if self.dir is None or not os.path.exists(self.dir):
-            log.error(f"Could not find directory: {self.dir}")
+        if self.dir_path is None or not os.path.exists(self.dir_path):
+            log.error(f"Could not find directory: {self.dir_path}")
             return False
-        main_nf = os.path.join(self.dir, "main.nf")
-        nf_config = os.path.join(self.dir, "nextflow.config")
+        main_nf = os.path.join(self.dir_path, "main.nf")
+        nf_config = os.path.join(self.dir_path, "nextflow.config")
         if not os.path.exists(main_nf) and not os.path.exists(nf_config):
-            if Path(self.dir).resolve().parts[-1].startswith("nf-core"):
-                raise UserWarning(f"Could not find a 'main.nf' or 'nextflow.config' file in '{self.dir}'")
-            log.warning(f"Could not find a 'main.nf' or 'nextflow.config' file in '{self.dir}'")
+            if Path(self.dir_path).resolve().parts[-1].startswith("nf-core"):
+                raise UserWarning(f"Could not find a 'main.nf' or 'nextflow.config' file in '{self.dir_path}'")
+            log.warning(f"Could not find a 'main.nf' or 'nextflow.config' file in '{self.dir_path}'")
         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")
+        modules_json_path = os.path.join(self.dir_path, "modules.json")
         if not os.path.exists(modules_json_path):
             log.info("Creating missing 'module.json' file.")
-            ModulesJson(self.dir).create()
+            ModulesJson(self.dir_path).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
 
@@ -112,7 +122,7 @@ def clear_component_dir(self, component_name, component_dir):
         try:
             shutil.rmtree(component_dir)
             # remove all empty directories
-            for dir_path, dir_names, filenames in os.walk(self.dir, topdown=False):
+            for dir_path, dir_names, filenames in os.walk(self.dir_path, topdown=False):
                 if not dir_names and not filenames:
                     try:
                         os.rmdir(dir_path)
@@ -127,7 +137,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
 
@@ -137,15 +147,17 @@ def components_from_repo(self, install_dir):
         Returns:
             [str]: The names of the modules/subworkflows
         """
-        repo_dir = Path(self.dir, self.component_type, install_dir)
+        repo_dir = Path(self.dir_path, self.component_type, install_dir)
         if not repo_dir.exists():
             raise LookupError(f"Nothing installed from {install_dir} in pipeline")
 
         return [
             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,18 +172,18 @@ 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
         the lint config from it
 
         Add parsed config to the `self.lint_config` class attribute.
         """
-        _, tools_config = nf_core.utils.load_tools_config(self.dir)
+        _, tools_config = nf_core.utils.load_tools_config(self.dir_path)
         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,10 +192,10 @@ def check_modules_structure(self):
             modules/nf-core/modules/TOOL/SUBTOOL
         """
         if self.repo_type == "pipeline":
-            wrong_location_modules = []
-            for directory, _, files in os.walk(Path(self.dir, "modules")):
+            wrong_location_modules: List[Path] = []
+            for directory, _, files in os.walk(Path(self.dir_path, "modules")):
                 if "main.nf" in files:
-                    module_path = Path(directory).relative_to(Path(self.dir, "modules"))
+                    module_path = Path(directory).relative_to(Path(self.dir_path, "modules"))
                     parts = module_path.parts
                     # Check that there are modules installed directly under the 'modules' directory
                     if parts[1] == "modules":
@@ -201,14 +213,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"))
+                shutil.rmtree(Path(self.dir_path, "modules", self.modules_repo.repo_path, "modules"))
                 # Regenerate modules.json file
-                modules_json = ModulesJson(self.dir)
+                modules_json = ModulesJson(self.dir_path)
                 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
         """
@@ -231,15 +243,15 @@ def check_patch_paths(self, patch_path, module_name):
                     for line in lines:
                         fh.write(line)
                 # Update path in modules.json if the file is in the correct format
-                modules_json = ModulesJson(self.dir)
+                modules_json = ModulesJson(self.dir_path)
                 modules_json.load()
                 if modules_json.has_git_url_and_modules():
                     modules_json.modules_json["repos"][self.modules_repo.remote_url]["modules"][
                         self.modules_repo.repo_path
-                    ][module_name]["patch"] = str(patch_path.relative_to(Path(self.dir).resolve()))
+                    ][module_name]["patch"] = str(patch_path.relative_to(Path(self.dir_path).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,9 +260,9 @@ 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")
+            workflow_files = Path(self.dir_path, "workflows").glob("*.nf")
             for workflow_file in workflow_files:
                 with open(workflow_file, "r") as fh:
                     # Check if component path is in the file using mmap

nf_core/components/components_utils.py

@@ -2,6 +2,7 @@
 import os
 import re
 from pathlib import Path
+from typing import Optional, Tuple, Union
 
 import questionary
 import rich.prompt
@@ -11,7 +12,9 @@
 log = logging.getLogger(__name__)
 
 
-def get_repo_info(directory, use_prompt=True):
+def get_repo_info(
+    directory: Union[str, Path], use_prompt=True
+) -> Tuple[Union[str, Path], Optional[str], Optional[str]]:
     """
     Determine whether this is a pipeline repository or a clone of
     nf-core/modules
@@ -21,7 +24,7 @@ def get_repo_info(directory, use_prompt=True):
         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: Path = 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)
@@ -77,7 +80,7 @@ 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):

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"