ruff auto-fixes

doc/source/conf.py

@@ -1,4 +1,4 @@
-#
+#  # noqa: INP001
 # Configuration file for the Sphinx documentation builder.
 #
 # This file does only contain a selection of the most common options. For a
@@ -23,7 +23,7 @@
 # -- Project information -----------------------------------------------------
 
 project = "QToolKit"
-copyright = "2023, Matgenix SRL"
+copyright = "2023, Matgenix SRL"  # noqa: A001
 author = "Guido Petretto, David Waroquiers"
 
 
@@ -109,7 +109,7 @@
     },
     "collapse_navigation": True,
     "announcement": (
-        "<p>" "QToolKit is still in beta phase. The API may change at any time." "</p>"
+        "<p>QToolKit is still in beta phase. The API may change at any time.</p>"
     ),
     # "navbar_end": ["theme-switcher", "navbar-icon-links"],
     # "navbar_end": ["theme-switcher", "version-switcher", "navbar-icon-links"],

src/qtoolkit/core/base.py

@@ -25,8 +25,8 @@ def _validate_monty(cls, __input_value):
         """
         try:
             super()._validate_monty(__input_value)
-        except ValueError as e:
+        except ValueError as exc:
             try:
                 return cls(__input_value)
             except Exception:
-                raise e
+                raise exc  # noqa: B904

src/qtoolkit/core/data_objects.py

@@ -2,11 +2,14 @@
 
 import abc
 from dataclasses import dataclass, fields
-from pathlib import Path
+from typing import TYPE_CHECKING
 
 from qtoolkit.core.base import QTKEnum, QTKObject
 from qtoolkit.core.exceptions import UnsupportedResourcesError
 
+if TYPE_CHECKING:
+    from pathlib import Path
+
 
 class SubmissionStatus(QTKEnum):
     SUCCESSFUL = "SUCCESSFUL"
@@ -190,7 +193,7 @@ class QResources(QTKObject):
     def __post_init__(self):
         if self.process_placement is None:
             if self.processes and not self.processes_per_node and not self.nodes:
-                self.process_placement = ProcessPlacement.NO_CONSTRAINTS  # type: ignore # due to QTKEnum
+                self.process_placement = ProcessPlacement.NO_CONSTRAINTS
             elif self.nodes and self.processes_per_node and not self.processes:
                 self.process_placement = ProcessPlacement.EVENLY_DISTRIBUTED
             elif not self._check_no_values():
@@ -203,18 +206,12 @@ def __post_init__(self):
         self.scheduler_kwargs = self.scheduler_kwargs or {}
 
     def _check_no_values(self) -> bool:
-        """
-        Check if all the attributes are None or empty.
-        """
-        for f in fields(self):
-            if self.__getattribute__(f.name):
-                return False
-
-        return True
+        """Check if all the attributes are None or empty."""
+        return all(not self.__getattribute__(f.name) for f in fields(self))
 
     def check_empty(self) -> bool:
         """
-        Check if the QResouces is empty and its content is coherent.
+        Check if the QResources is empty and its content is coherent.
         Raises an error if process_placement is None, but some attributes are set.
         """
         if self.process_placement is not None:

src/qtoolkit/core/exceptions.py

@@ -1,7 +1,5 @@
 class QTKException(Exception):
-    """
-    Base class for all the exceptions generated by qtoolkit.
-    """
+    """Base class for all the exceptions generated by qtoolkit."""
 
 
 class CommandFailedError(QTKException):

src/qtoolkit/host/base.py

@@ -2,10 +2,13 @@
 
 import abc
 from dataclasses import dataclass
-from pathlib import Path
+from typing import TYPE_CHECKING
 
 from qtoolkit.core.base import QTKObject
 
+if TYPE_CHECKING:
+    from pathlib import Path
+
 
 @dataclass
 class HostConfig(QTKObject):
@@ -30,7 +33,7 @@ def execute(
         # stdout=None,
         # stderr=None,
     ):
-        """Execute the given command on the host
+        """Execute the given command on the host.
 
         Parameters
         ----------

src/qtoolkit/host/local.py

@@ -11,7 +11,7 @@ class LocalHost(BaseHost):
     # def __init__(self, config):
     #     self.config = config
     def execute(self, command: str | list[str], workdir: str | Path | None = None):
-        """Execute the given command on the host
+        """Execute the given command on the host.
 
         Note that the command is executed with shell=True, so commands can
         be exposed to command injection. Consider whether to escape part of

src/qtoolkit/host/remote.py

@@ -2,12 +2,15 @@
 
 import io
 from dataclasses import dataclass, field
-from pathlib import Path
+from typing import TYPE_CHECKING
 
 import fabric
 
 from qtoolkit.host.base import BaseHost, HostConfig
 
+if TYPE_CHECKING:
+    from pathlib import Path
+
 # from fabric import Connection, Config
 
 
@@ -128,7 +131,7 @@ class RemoteConfig(HostConfig):
 class RemoteHost(BaseHost):
     """
     Execute commands on a remote host.
-    For some commands assumes the remote can run unix
+    For some commands assumes the remote can run unix.
     """
 
     def __init__(self, config: RemoteConfig):
@@ -143,7 +146,7 @@ def connection(self):
         return self._connection
 
     def execute(self, command: str | list[str], workdir: str | Path | None = None):
-        """Execute the given command on the host
+        """Execute the given command on the host.
 
         Parameters
         ----------
@@ -161,18 +164,14 @@ def execute(self, command: str | list[str], workdir: str | Path | None = None):
         exit_code : int
             Exit code of the command.
         """
-
         if isinstance(command, (list, tuple)):
             command = " ".join(command)
 
         # TODO: check here if we use the context manager. What happens if we provide the
         #  connection from outside (not through a config) and we want to keep it alive ?
 
         # TODO: check if this works:
-        if not workdir:
-            workdir = "."
-        else:
-            workdir = str(workdir)
+        workdir = str(workdir) if workdir else "."
         with self.connection.cd(workdir):
             out = self.connection.run(command, hide=True, warn=True)
 
@@ -185,10 +184,11 @@ def mkdir(self, directory, recursive: bool = True, exist_ok: bool = True) -> boo
             command += "-p "
         command += str(directory)
         try:
-            stdout, stderr, returncode = self.execute(command)
-            return returncode == 0
+            _stdout, _stderr, returncode = self.execute(command)
         except Exception:
             return False
+        else:
+            return returncode == 0
 
     def write_text_file(self, filepath, content):
         """Write content to a file on the host."""

src/qtoolkit/io/base.py

@@ -3,13 +3,16 @@
 import abc
 import shlex
 from dataclasses import fields
-from pathlib import Path
 from string import Template
+from typing import TYPE_CHECKING
 
 from qtoolkit.core.base import QTKObject
 from qtoolkit.core.data_objects import CancelResult, QJob, QResources, SubmissionResult
 from qtoolkit.core.exceptions import UnsupportedResourcesError
 
+if TYPE_CHECKING:
+    from pathlib import Path
+
 
 class QTemplate(Template):
     delimiter = "$$"
@@ -74,9 +77,8 @@ def generate_header(self, options: dict | QResources | None) -> str:
 
         options = options or {}
 
-        if isinstance(options, QResources):
-            if not options.check_empty():
-                options = self.check_convert_qresources(options)
+        if isinstance(options, QResources) and not options.check_empty():
+            options = self.check_convert_qresources(options)
 
         template = QTemplate(self.header_template)
 
@@ -89,10 +91,7 @@ def generate_header(self, options: dict | QResources | None) -> str:
 
         unclean_header = template.safe_substitute(options)
         # Remove lines with leftover $$.
-        clean_header = []
-        for line in unclean_header.split("\n"):
-            if "$$" not in line:
-                clean_header.append(line)
+        clean_header = [line for line in unclean_header.split("\n") if "$$" not in line]
 
         return "\n".join(clean_header)
 

src/qtoolkit/io/pbs.py

@@ -2,6 +2,7 @@
 
 import re
 from datetime import timedelta
+from typing import ClassVar
 
 from qtoolkit.core.data_objects import (
     CancelResult,
@@ -156,9 +157,7 @@ def parse_cancel_output(self, exit_code, stdout, stderr) -> CancelResult:
         )
 
     def _get_job_cmd(self, job_id: str):
-        cmd = f"qstat -f {job_id}"
-
-        return cmd
+        return f"qstat -f {job_id}"
 
     def parse_job_output(self, exit_code, stdout, stderr) -> QJob | None:
         out = self.parse_jobs_list_output(exit_code, stdout, stderr)
@@ -224,7 +223,7 @@ def parse_jobs_list_output(self, exit_code, stdout, stderr) -> list[QJob]:
 
         jobs_list = []
         for chunk in jobs_chunks:
-            chunk = chunk.strip()
+            chunk = chunk.strip()  # noqa: PLW2901
             if not chunk:
                 continue
 
@@ -243,9 +242,9 @@ def parse_jobs_list_output(self, exit_code, stdout, stderr) -> list[QJob]:
 
             try:
                 pbs_job_state = PBSState(job_state_string)
-            except ValueError:
+            except ValueError as exc:
                 msg = f"Unknown job state {job_state_string} for job id {qjob.job_id}"
-                raise OutputParsingError(msg)
+                raise OutputParsingError(msg) from exc
             qjob.sub_state = pbs_job_state
             qjob.state = pbs_job_state.qstate
 
@@ -299,7 +298,6 @@ def _convert_str_to_time(time_str: str | None):
         Convert a string in the format used by PBS DD:HH:MM:SS to a number of seconds.
         It may contain only H:M:S, only M:S or only S.
         """
-
         if not time_str:
             return None
 
@@ -312,8 +310,8 @@ def _convert_str_to_time(time_str: str | None):
             for i, v in enumerate(reversed(time_split)):
                 time[i] = int(v)
 
-        except ValueError:
-            raise OutputParsingError()
+        except ValueError as exc:
+            raise OutputParsingError from exc
 
         return time[3] * 86400 + time[2] * 3600 + time[1] * 60 + time[0]
 
@@ -335,14 +333,14 @@ def _convert_memory_str(memory: str | None) -> int | None:
             raise OutputParsingError(f"Unknown units {units}")
         try:
             v = int(memory)
-        except ValueError:
-            raise OutputParsingError
+        except ValueError as exc:
+            raise OutputParsingError from exc
 
         return v * (1024 ** power_labels[units])
 
     # helper attribute to match the values defined in QResources and
     # the dictionary that should be passed to the template
-    _qresources_mapping = {
+    _qresources_mapping: ClassVar = {
         "queue_name": "queue",
         "job_name": "job_name",
         "account": "account",
@@ -353,22 +351,20 @@ def _convert_memory_str(memory: str | None) -> int | None:
     }
 
     @staticmethod
-    def _convert_time_to_str(time: int | float | timedelta) -> str:
+    def _convert_time_to_str(time: int | float | timedelta) -> str:  # noqa: PYI041
         if not isinstance(time, timedelta):
             time = timedelta(seconds=time)
 
         hours, remainder = divmod(int(time.total_seconds()), 3600)
         minutes, seconds = divmod(remainder, 60)
 
-        time_str = f"{hours}:{minutes}:{seconds}"
-        return time_str
+        return f"{hours}:{minutes}:{seconds}"
 
     def _convert_qresources(self, resources: QResources) -> dict:
         """
         Converts a QResources instance to a dict that will be used to fill in the
         header of the submission script.
         """
-
         header_dict = {}
         for qr_field, pbs_field in self._qresources_mapping.items():
             val = getattr(resources, qr_field)

src/qtoolkit/io/shell.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from pathlib import Path
+from typing import TYPE_CHECKING
 
 from qtoolkit.core.data_objects import (
     CancelResult,
@@ -19,6 +19,9 @@
 )
 from qtoolkit.io.base import BaseSchedulerIO
 
+if TYPE_CHECKING:
+    from pathlib import Path
+
 # States in from ps command, extracted from man ps.
 # D    uninterruptible sleep (usually IO)
 # R    running or runnable (on run queue)
@@ -152,9 +155,7 @@ def parse_cancel_output(self, exit_code, stdout, stderr) -> CancelResult:
         )
 
     def _get_job_cmd(self, job_id: str):
-        cmd = self._get_jobs_list_cmd(job_ids=[job_id])
-
-        return cmd
+        return self._get_jobs_list_cmd(job_ids=[job_id])
 
     def parse_job_output(self, exit_code, stdout, stderr) -> QJob | None:
         """Parse the output of the ps command and return the corresponding QJob object.
@@ -239,9 +240,9 @@ def parse_jobs_list_output(self, exit_code, stdout, stderr) -> list[QJob]:
 
             try:
                 shell_job_state = ShellState(data[3][0])
-            except ValueError:
+            except ValueError as exc:
                 msg = f"Unknown job state {data[3]} for job id {qjob.job_id}"
-                raise OutputParsingError(msg)
+                raise OutputParsingError(msg) from exc
             qjob.sub_state = shell_job_state
             qjob.state = shell_job_state.qstate
 
@@ -276,7 +277,6 @@ def _convert_str_to_time(time_str: str | None) -> int | None:
         Convert a string in the format used in etime [[DD-]hh:]mm:ss to a
         number of seconds.
         """
-
         if not time_str:
             return None
 
@@ -295,9 +295,9 @@ def _convert_str_to_time(time_str: str | None) -> int | None:
             elif len(time_split) == 2:
                 minutes, seconds = (int(v) for v in time_split)
             else:
-                raise OutputParsingError()
+                raise OutputParsingError
 
-        except ValueError:
-            raise OutputParsingError()
+        except ValueError as exc:
+            raise OutputParsingError from exc
 
         return days * 86400 + hours * 3600 + minutes * 60 + seconds