Add more ruff rules (B, PTH, RSE, SIM, C4, BLE)

.github/workflows/changelog.py

@@ -135,9 +135,7 @@ def _skip_existing_entry_for_this_pr(line: str, same_section: bool = True) -> st
     # If the line already contains a link to the PR, don't add it again.
     line = _skip_existing_entry_for_this_pr(line, same_section=False)
 
-    if (
-        line.startswith("## ") and not line.strip() == "# nf-core/tools: Changelog"
-    ):  # Version header, e.g. "## v2.12dev"
+    if line.startswith("## ") and line.strip() != "# nf-core/tools: Changelog":  # Version header, e.g. "## v2.12dev"
         print(f"Found version header: {line.strip()}")
         updated_lines.append(line)
 

CHANGELOG.md

@@ -5,6 +5,7 @@
 ### General
 
 - Lock file maintenance ([#4044](https://github.com/nf-core/tools/pull/4044))
+- Add more ruff rules (B, PTH) ([#4034](https://github.com/nf-core/tools/pull/4034))
 
 ### Linting
 

docs/api/_src/conf.py

@@ -11,12 +11,12 @@
 # add these directories to sys.path here. If the directory is relative to the
 # documentation root, use os.path.abspath to make it absolute, like shown here.
 #
-import os
 import sys
+from pathlib import Path
 
 import nf_core
 
-sys.path.insert(0, os.path.abspath("../../../nf_core"))
+sys.path.insert(0, str(Path("../../../nf_core").resolve()))
 
 # -- Project information -----------------------------------------------------
 

nf_core/__main__.py

@@ -6,6 +6,7 @@
 import sys
 from pathlib import Path
 
+import requests
 import rich
 import rich.console
 import rich.logging
@@ -123,15 +124,15 @@ def run_nf_core():
                     f"[bold bright_yellow]    There is a new version of nf-core/tools available! ({remote_vers})",
                     highlight=False,
                 )
-        except Exception as e:
+        except requests.exceptions.RequestException as e:
             log.debug(f"Could not check latest version: {e}")
         stderr.print("\n")
     # Launch the click cli
     nf_core_cli(auto_envvar_prefix="NFCORE")
 
 
 @tui(command="interface", help="Launch the nf-core interface")
-@click.group(context_settings=dict(help_option_names=["-h", "--help"]))
+@click.group(context_settings={"help_option_names": ["-h", "--help"]})
 @click.version_option(__version__)
 @click.option(
     "-v",
@@ -249,7 +250,7 @@ def command_pipelines_create(ctx, name, description, author, version, force, out
     is_flag=True,
     default=Path(os.environ.get("GITHUB_REF", "").strip(" '\"")).parent.name in ["master", "main"]
     and os.environ.get("GITHUB_REPOSITORY", "").startswith("nf-core/")
-    and not os.environ.get("GITHUB_REPOSITORY", "") == "nf-core/tools",
+    and os.environ.get("GITHUB_REPOSITORY", "") != "nf-core/tools",
     help="Execute additional checks for release-ready workflows.",
 )
 @click.option(
@@ -485,7 +486,7 @@ def command_pipelines_create_params_file(ctx, pipeline, revision, output, force,
     "-o",
     "--params-out",
     type=click.Path(),
-    default=os.path.join(os.getcwd(), "nf-params.json"),
+    default=str(Path.cwd() / "nf-params.json"),
     help="Path to save run parameters file",
 )
 @click.option(
@@ -2220,7 +2221,7 @@ def command_list(ctx, keywords, sort, json, show_archived):
     "-o",
     "--params-out",
     type=click.Path(),
-    default=os.path.join(os.getcwd(), "nf-params.json"),
+    default=str(Path.cwd() / "nf-params.json"),
     help="Path to save run parameters file",
 )
 @click.option(
@@ -2433,7 +2434,7 @@ def command_download(
     is_flag=True,
     default=Path(os.environ.get("GITHUB_REF", "").strip(" '\"")).parent.name in ["master", "main"]
     and os.environ.get("GITHUB_REPOSITORY", "").startswith("nf-core/")
-    and not os.environ.get("GITHUB_REPOSITORY", "") == "nf-core/tools",
+    and os.environ.get("GITHUB_REPOSITORY", "") != "nf-core/tools",
     help="Execute additional checks for release-ready workflows.",
 )
 @click.option(

nf_core/commands_pipelines.py

@@ -1,5 +1,4 @@
 import logging
-import os
 import sys
 from pathlib import Path
 
@@ -448,7 +447,7 @@ def pipelines_schema_docs(schema_path, output, format, force, columns):
     """
     Outputs parameter documentation for a pipeline schema.
     """
-    if not os.path.exists(schema_path):
+    if not schema_path.exists():
         log.error("Could not find 'nextflow_schema.json' in current directory. Please specify a path.")
         sys.exit(1)
 

nf_core/components/components_command.py

@@ -50,7 +50,7 @@ def _configure_repo_and_paths(self, nf_dir_req: bool = True) -> None:
         """
         try:
             if self.directory:
-                if self.directory == Path(".") and not nf_dir_req:
+                if self.directory == Path() and not nf_dir_req:
                     self.no_prompts = True
                 self.directory, self.repo_type, self.org = get_repo_info(self.directory, use_prompt=not self.no_prompts)
         except UserWarning:
@@ -69,6 +69,7 @@ def get_local_components(self) -> list[str]:
         Get the local modules/subworkflows in a pipeline
         """
         local_component_dir = Path(self.directory, self.component_type, "local")
+        # TODO: Return list of Path objects instead of strings to avoid unnecessary conversion
         return [
             str(Path(directory).relative_to(local_component_dir))
             for directory, _, files in os.walk(local_component_dir)
@@ -85,6 +86,7 @@ def get_components_clone_modules(self) -> list[str]:
             component_base_path = Path(self.directory, self.default_modules_path)
         elif self.component_type == "subworkflows":
             component_base_path = Path(self.directory, self.default_subworkflows_path)
+        # TODO: Return list of Path objects instead of strings to avoid unnecessary conversion
         return [
             str(Path(directory).relative_to(component_base_path))
             for directory, _, files in os.walk(component_base_path)
@@ -155,6 +157,7 @@ def components_from_repo(self, install_dir: str) -> list[str]:
         if not repo_dir.exists():
             raise LookupError(f"Nothing installed from {install_dir} in pipeline")
 
+        # TODO: Return list of Path objects instead of strings to avoid unnecessary conversion
         return [
             str(Path(dir_path).relative_to(repo_dir)) for dir_path, _, files in os.walk(repo_dir) if "main.nf" in files
         ]
@@ -257,6 +260,7 @@ def check_patch_paths(self, patch_path: Path, module_name: str) -> None:
                     and self.modules_repo.repo_path is not None
                     and modules_json.modules_json is not None
                 ):
+                    # TODO: Consider if this needs to be a string or if Path object would work
                     modules_json.modules_json["repos"][self.modules_repo.remote_url]["modules"][
                         self.modules_repo.repo_path
                     ][module_name]["patch"] = str(patch_path.relative_to(self.directory.resolve()))
@@ -275,18 +279,15 @@ def check_if_in_include_stmts(self, component_path: str) -> dict[str, list[dict[
         if self.repo_type == "pipeline":
             workflow_files = Path(self.directory, "workflows").glob("*.nf")
             for workflow_file in workflow_files:
-                with open(workflow_file) as fh:
+                with open(workflow_file) as fh, mmap.mmap(fh.fileno(), 0, access=mmap.ACCESS_READ) as s:
                     # Check if component path is in the file using mmap
-                    with mmap.mmap(fh.fileno(), 0, access=mmap.ACCESS_READ) as s:
-                        if s.find(component_path.encode()) != -1:
-                            # If the component path is in the file, check for include statements
-                            for i, line in enumerate(fh):
-                                if line.startswith("include") and component_path in line:
-                                    if str(workflow_file) not in include_stmts:
-                                        include_stmts[str(workflow_file)] = []
-                                    include_stmts[str(workflow_file)].append(
-                                        {"line_number": i + 1, "line": line.rstrip()}
-                                    )
+                    if s.find(component_path.encode()) != -1:
+                        # If the component path is in the file, check for include statements
+                        for i, line in enumerate(fh):
+                            if line.startswith("include") and component_path in line:
+                                if str(workflow_file) not in include_stmts:
+                                    include_stmts[str(workflow_file)] = []
+                                include_stmts[str(workflow_file)].append({"line_number": i + 1, "line": line.rstrip()})
 
             return include_stmts
         else:

nf_core/components/components_completion.py

@@ -1,5 +1,6 @@
 import sys
 
+import git
 from click.shell_completion import CompletionItem
 
 from nf_core.modules.list import ModuleList
@@ -24,7 +25,7 @@ def autocomplete_components(ctx, param, incomplete: str, component_type: str, li
         available_components = components_list.modules_repo.get_avail_components(component_type)
 
         return [CompletionItem(comp) for comp in available_components if comp.startswith(incomplete)]
-    except Exception as e:
+    except (git.exc.GitError, LookupError, OSError) as e:
         print(f"[ERROR] Autocomplete failed: {e}", file=sys.stderr)
         return []
 

nf_core/components/components_differ.py

@@ -187,7 +187,7 @@ def write_diff_file(
                 elif diff_status == ComponentsDiffer.DiffEnum.REMOVED:
                     # The file was removed between the commits
                     fh.write(f"'{Path(dsp_from_dir, file)}' was removed\n")
-                elif limit_output and not file.suffix == ".nf":
+                elif limit_output and file.suffix != ".nf":
                     # Skip printing the diff for files other than main.nf
                     fh.write(f"Changes in '{Path(component, file)}' but not shown\n")
                 else:
@@ -286,7 +286,7 @@ def print_diff(
             elif diff_status == ComponentsDiffer.DiffEnum.REMOVED:
                 # The file was removed between the commits
                 log.info(f"'{Path(dsp_from_dir, file)}' was removed")
-            elif limit_output and not file.suffix == ".nf":
+            elif limit_output and file.suffix != ".nf":
                 # Skip printing the diff for files other than main.nf
                 log.info(f"Changes in '{Path(component, file)}' but not shown")
             else:

nf_core/components/components_utils.py

@@ -247,10 +247,10 @@ def get_channel_info_from_biotools(
     inputs = {}
     outputs = {}
 
-    def _iterate_input_output(type) -> DictWithStrAndTuple:
+    def _iterate_input_output(funct_data, type) -> DictWithStrAndTuple:
         type_info = {}
-        if type in funct:
-            for element in funct[type]:
+        if type in funct_data:
+            for element in funct_data[type]:
                 if "data" in element:
                     element_name = "_".join(element["data"]["term"].lower().split(" "))
                     uris = [element["data"]["uri"]]
@@ -272,8 +272,8 @@ def _iterate_input_output(type) -> DictWithStrAndTuple:
             if "function" in tool:
                 # Parse all tool functions
                 for funct in tool["function"]:
-                    inputs.update(_iterate_input_output("input"))
-                    outputs.update(_iterate_input_output("output"))
+                    inputs.update(_iterate_input_output(funct, "input"))
+                    outputs.update(_iterate_input_output(funct, "output"))
             return inputs, outputs
 
     # If the tool name was not found in the response

nf_core/components/create.py

@@ -2,7 +2,6 @@
 The ComponentCreate class handles generating of module and subworkflow templates
 """
 
-import glob
 import json
 import logging
 import re
@@ -38,7 +37,7 @@ class ComponentCreate(ComponentCommand):
     def __init__(
         self,
         component_type: str,
-        directory: Path = Path("."),
+        directory: Path = Path(),
         component: str = "",
         author: str | None = None,
         process_label: str | None = None,
@@ -403,7 +402,7 @@ def _get_component_dirs(self) -> dict[str, Path]:
                 )
 
             # If no subtool, check that there isn't already a tool/subtool
-            tool_glob = glob.glob(f"{Path(self.directory, self.component_type, self.org, self.component)}/*/main.nf")
+            tool_glob = list(Path(self.directory, self.component_type, self.org, self.component).glob("*/main.nf"))
             if not self.subtool and tool_glob and not self.migrate_pytest:
                 raise UserWarning(
                     f"Module subtool '{tool_glob[0]}' exists already, cannot make tool '{self.component_name}'"
@@ -427,7 +426,7 @@ def _get_username(self):
         try:
             gh_auth_user = json.loads(subprocess.check_output(["gh", "api", "/user"], stderr=subprocess.DEVNULL))
             author_default = f"@{gh_auth_user['login']}"
-        except Exception as e:
+        except (subprocess.CalledProcessError, FileNotFoundError) as e:
             log.debug(f"Could not find GitHub username using 'gh' cli command: [red]{e}")
 
         # Regex to valid GitHub username: https://github.com/shinnn/github-username-regex
@@ -626,7 +625,7 @@ def generate_meta_yml_file(self) -> None:
 
             if hasattr(self, "outputs") and len(self.outputs) > 0:
                 outputs_dict: dict[str, list | dict] = {}
-                for i, (output_name, ontologies) in enumerate(self.outputs.items()):
+                for _i, (output_name, ontologies) in enumerate(self.outputs.items()):
                     channel_contents: list[list[dict] | dict] = []
                     if self.has_meta:
                         channel_contents.append(

nf_core/components/info.py

@@ -1,5 +1,4 @@
 import logging
-import os
 from pathlib import Path
 
 import questionary
@@ -200,9 +199,9 @@ def get_local_yaml(self) -> dict | None:
             log.debug(f"{self.component_type[:-1].title()} '{self.component}' meta.yml not found locally")
         else:
             component_base_path = Path(self.directory, self.component_type, self.org)
-            if self.component in os.listdir(component_base_path):
-                comp_dir = Path(component_base_path, self.component)
-                meta_fn = Path(comp_dir, "meta.yml")
+            comp_dir = component_base_path / self.component
+            if comp_dir.is_dir():
+                meta_fn = comp_dir / "meta.yml"
                 if meta_fn.exists():
                     log.debug(f"Found local file: {meta_fn}")
                     with open(meta_fn) as fh:
@@ -375,7 +374,7 @@ def generate_component_info_help(self):
             )
             if self.component_type == "subworkflows":
                 subworkflow_config = Path(install_folder, self.component, "nextflow.config").relative_to(self.directory)
-                if os.path.isfile(subworkflow_config):
+                if subworkflow_config.exists():
                     renderables.append(
                         Text.from_markup("\n [blue]Add the following config statement to use this subworkflow:")
                     )

nf_core/components/install.py

@@ -1,5 +1,4 @@
 import logging
-import os
 from pathlib import Path
 
 import questionary
@@ -248,10 +247,11 @@ def collect_and_verify_name(
 
             raise ValueError
 
-        if self.current_remote.remote_url == modules_repo.remote_url:
-            if not modules_repo.component_exists(component, self.component_type, commit=self.current_sha):
-                warn_msg = f"{self.component_type[:-1].title()} '{component}' not found in remote '{modules_repo.remote_url}' ({modules_repo.branch})"
-                log.warning(warn_msg)
+        if self.current_remote.remote_url == modules_repo.remote_url and not modules_repo.component_exists(
+            component, self.component_type, commit=self.current_sha
+        ):
+            warn_msg = f"{self.component_type[:-1].title()} '{component}' not found in remote '{modules_repo.remote_url}' ({modules_repo.branch})"
+            log.warning(warn_msg)
 
         return component
 
@@ -263,7 +263,7 @@ def check_component_installed(self, component, current_version, component_dir, m
             True: if the component is not installed
             False: if the component is installed
         """
-        if (current_version is not None and os.path.exists(component_dir)) and not force:
+        if (current_version is not None and component_dir.exists()) and not force:
             # make sure included components are also installed
             if self.component_type == "subworkflows":
                 self.install_included_components(component_dir)
@@ -343,9 +343,9 @@ def check_alternate_remotes(self, modules_json):
             False: if problematic components are not found
         """
         modules_json.load()
-        for repo_url, repo_content in modules_json.modules_json.get("repos", dict()).items():
+        for repo_url, repo_content in modules_json.modules_json.get("repos", {}).items():
             for component_type in repo_content:
-                for directory in repo_content.get(component_type, dict()).keys():
+                for directory in repo_content.get(component_type, {}):
                     if directory == self.modules_repo.repo_path and repo_url != self.modules_repo.remote_url:
                         return True
         return False

nf_core/components/lint/__init__.py

@@ -259,7 +259,7 @@ def _print_results(self, show_passed=False, sort_by="test", plain_text=False):
             try:
                 for lint_result in tests:
                     max_name_len = max(len(lint_result.component_name), max_name_len)
-            except Exception:
+            except (AttributeError, TypeError):
                 pass
 
         # Helper function to format test links nicely
@@ -285,8 +285,9 @@ def format_result(test_results: list[LintResult], table: Table) -> Table:
                     module_name = lint_result.component_name
 
                 # Make the filename clickable to open in VSCode
-                file_path = os.path.relpath(lint_result.file_path, self.directory)
-                file_path_link = f"[link=vscode://file/{os.path.abspath(file_path)}]{file_path}[/link]"
+                file_path_obj = Path(lint_result.file_path)
+                file_path_rel = file_path_obj.relative_to(self.directory)
+                file_path_link = f"[link=vscode://file/{file_path_obj.resolve()}]{file_path_rel}[/link]"
 
                 # Add link to the test documentation
                 tools_version = __version__