Feature/python typing/components command.py

.github/workflows/lint-code.yml

@@ -85,3 +85,17 @@ jobs:
  with:
  isortVersion: "latest"
  requirementsFiles: "requirements.txt requirements-dev.txt"
+
+ mypy:
+ runs-on: ubuntu-latest
+ steps:
+ - name: Setup Python
+ uses: actions/setup-python@v4
+ with:
+ python-version: 3.8
+ - name: Checkout
+ uses: actions/checkout@v3
+ - name: Install mypy
+ run: pip install mypy
+ - name: Run mypy
+ run: mypy nf-core/

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
 
@@ -19,17 +20,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: Optional[Union[str, Path]],
+ 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.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.
@@ -48,21 +57,21 @@ 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 = Path(self.dir, 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
  """
@@ -76,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
@@ -91,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
 
@@ -126,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
 
@@ -136,15 +145,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 = Path(self.dir, 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
 
@@ -159,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.
 
  Look for a file called either `.nf-core-lint.yml` or
@@ -181,7 +192,7 @@ def load_lint_config(self):
  except FileNotFoundError:
  log.debug(f"No lint config file found: {config_fn}")
 
- 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
@@ -190,11 +201,11 @@ 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"))
- parts = module_path.parts
+ module_path: Path = Path(directory).relative_to(Path(self.dir, "modules"))
+ parts: Tuple[str, ...] = module_path.parts
  # Check that there are modules installed directly under the 'modules' directory
  if parts[1] == "modules":
  wrong_location_modules.append(module_path)
@@ -208,17 +219,17 @@ def check_modules_structure(self):
  )
  # Move wrong modules to the right directory
  for module in wrong_location_modules:
- 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)
+ modules_dir: Path = Path("modules").resolve()
+ correct_dir: Path = Path(modules_dir, self.modules_repo.repo_path, Path(*module.parts[2:]))
+ wrong_dir: Path = Path(modules_dir, module)
  shutil.move(wrong_dir, 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: ModulesJson = 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
  """
@@ -241,15 +252,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 = ModulesJson(self.dir)
  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()))
  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:
@@ -258,9 +269,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 = Path(self.dir, "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
@@ -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,7 @@ norecursedirs = [ ".*", "build", "dist", "*.egg", "data", "__pycache__", ".githu
 profile = "black"
 known_first_party = ["nf_core"]
 multi_line_output = 3
+
+[mypy]
+ignore_missing_imports = true
+follow_imports = "skip"