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"