Merge pull request #4 from gpetretto/develop

typing

src/qtoolkit/core/data_objects.py

@@ -1,5 +1,6 @@
+from __future__ import annotations
+
 from dataclasses import dataclass
-from typing import List, Optional, Union
 
 from qtoolkit.core.base import QBase, QEnum
 
@@ -12,12 +13,12 @@ class SubmissionStatus(QEnum):
 
 @dataclass
 class SubmissionResult(QBase):
-    job_id: Optional[Union[int, str]] = None
-    step_id: Optional[int] = None
-    exit_code: Optional[int] = None
-    stdout: Optional[str] = None
-    stderr: Optional[str] = None
-    status: Optional[SubmissionStatus] = None
+    job_id: int | str | None = None
+    step_id: int | None = None
+    exit_code: int | None = None
+    stdout: str | None = None
+    stderr: str | None = None
+    status: SubmissionStatus | None = None
 
 
 class CancelStatus(QEnum):
@@ -28,12 +29,12 @@ class CancelStatus(QEnum):
 
 @dataclass
 class CancelResult(QBase):
-    job_id: Optional[Union[int, str]] = None
-    step_id: Optional[int] = None
-    exit_code: Optional[int] = None
-    stdout: Optional[str] = None
-    stderr: Optional[str] = None
-    status: Optional[CancelStatus] = None
+    job_id: int | str | None = None
+    step_id: int | None = None
+    exit_code: int | None = None
+    stdout: str | None = None
+    stderr: str | None = None
+    status: CancelStatus | None = None
 
 
 class QState(QEnum):
@@ -111,7 +112,7 @@ class QResources(QBase):
 
     queue_name: str = None
     memory: int = 1024
-    nodes: Union[int, List] = 1
+    nodes: int | list = 1
     cpus_per_node: int = 1
     cores_per_cpu: int = 1
     hyperthreading: int = 1
@@ -135,10 +136,10 @@ class QOptions(QBase):
 
 @dataclass
 class QJob(QBase):
-    name: Optional[str] = None
-    qid: Optional[str] = None
-    exit_status: Optional[int] = None
-    state: Optional[QState] = None  # Standard
-    sub_state: Optional[QSubState] = None
-    resources: Optional[QResources] = None
-    job_info: Optional[QJobInfo] = None
+    name: str | None = None
+    qid: str | None = None
+    exit_status: int | None = None
+    state: QState | None = None  # Standard
+    sub_state: QSubState | None = None
+    resources: QResources | None = None
+    job_info: QJobInfo | None = None

src/qtoolkit/core/jobs.py

@@ -1,19 +1,20 @@
+# from __future__ import annotations
+#
 # from dataclasses import dataclass
 #
 # from qtoolkit.core.base import QBase
 # from qtoolkit.core.data_objects import QState, QSubState, QResources, QJobInfo
 #
-# from typing import Optional
 #
 # TODO: this has been moved to data_objects for now. See if it should be here
 #  for some reason ?
 # @dataclass
 # class QJob(QBase):
 #     name: str
-#     qid: Optional[str]
-#     exit_status: Optional[int]
-#     state: Optional[QState]  # Standard
-#     sub_state: Optional[QSubState]
-#     queue: Optional[str]
-#     resources: Optional[QResources]
-#     job_info: Optional[QJobInfo]
+#     qid: str | None
+#     exit_status: int | None
+#     state: QState | None  # Standard
+#     sub_state: QSubState | None
+#     queue: str | None
+#     resources: QResources | None
+#     job_info: QJobInfo | None

src/qtoolkit/host/base.py

@@ -1,27 +1,28 @@
+from __future__ import annotations
+
 import abc
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Union
 
 from qtoolkit.core.base import QBase
 
 
 @dataclass
 class HostConfig(QBase):
-    root_dir: Union[str, Path]
+    root_dir: str | Path
 
 
 class BaseHost(QBase):
     """Base Host class."""
 
     # def __init__(self, config, user):
-    def __init__(self, config=None):
-        self.config = config or {}
+    def __init__(self, config: HostConfig | None = None) -> None:
+        self.config = config
 
     #     self.user = user
 
     @abc.abstractmethod
-    def execute(self, command, stdin=None, stdout=None, stderr=None):
+    def execute(self, command: str | list[str], stdin=None, stdout=None, stderr=None):
         """Execute the given command on the host
 
         Parameters
@@ -44,6 +45,6 @@ def execute(self, command, stdin=None, stdout=None, stderr=None):
         raise NotImplementedError
 
     @abc.abstractmethod
-    def mkdir(self, directory, recursive=True, exist_ok=True):
+    def mkdir(self, directory, recursive: bool = True, exist_ok: bool = True):
         """Create directory on the host."""
         raise NotImplementedError

src/qtoolkit/host/local.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import subprocess
 from pathlib import Path
 
@@ -29,8 +31,8 @@ def execute(self, command):
         proc = subprocess.run(command, capture_output=True)
         return proc.stdout.decode(), proc.stderr.decode(), proc.returncode
 
-    def mkdir(self, directory, recursive=True, exist_ok=True):
+    def mkdir(self, directory, recursive=True, exist_ok=True) -> None:
         Path(directory).mkdir(parents=recursive, exist_ok=exist_ok)
 
-    def write_file(self, filepath, content):
+    def write_file(self, filepath, content) -> None:
         Path(filepath).write_text(content)

src/qtoolkit/host/remote.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from qtoolkit.host.base import BaseHost
 
 

src/qtoolkit/queue/base.py

@@ -1,8 +1,9 @@
+from __future__ import annotations
+
 import abc
 from dataclasses import dataclass, field
 from pathlib import Path
 from string import Template
-from typing import List, Optional, Union
 
 from qtoolkit.core.base import QBase
 from qtoolkit.core.data_objects import QJob
@@ -28,13 +29,11 @@ class BaseQueue(QBase):
 
     header_template: str
     name: str = "name of queue"
-    host: Union[
-        BaseHost,
-    ] = field(default_factory=LocalHost)
+    host: BaseHost = field(default_factory=LocalHost)
     default_shebang: str = "#!/bin/bash"
 
     SCRIPT_FNAME = "submit.script"
-    SUBMIT_CMD: Optional[str] = None
+    SUBMIT_CMD: str | None = None
 
     # host : QToolKit.Host or paramiko Client or Fabric client or None
     #         The host where the command should be executed.
@@ -113,7 +112,7 @@ def execute_cmd(self, cmd):
 
     def get_submission_script(
         self,
-        commands: Union[str, List[str]],
+        commands: str | list[str] | None,
         resources=None,
         submit_dir=None,
         environment=None,
@@ -291,7 +290,7 @@ def get_submit_cmd(self, script_file: str = SCRIPT_FNAME) -> str:
 
         return f"{self.SUBMIT_CMD} {script_file}"
 
-    def get_cancel_cmd(self, job: Union[QJob, int, str]) -> str:
+    def get_cancel_cmd(self, job: QJob | int | str) -> str:
         """
         Get the command used to cancel a given job.
 
@@ -302,7 +301,7 @@ def get_cancel_cmd(self, job: Union[QJob, int, str]) -> str:
         job_id = QJob.qid if isinstance(job, QJob) else job
         return f"{self.CANCEL_CMD} {job_id}"
 
-    def write_script(self, script_fpath: Union[str, Path], script_content: str):
+    def write_script(self, script_fpath: str | Path, script_content: str) -> None:
         self.host.write_file(script_fpath, script_content)
 
     @abc.abstractmethod
@@ -315,7 +314,7 @@ def _parse_cancel_cmd_output(self, exit_code, stdout, stderr):
 
     def submit(
         self,
-        commands: Union[str, List[str]],
+        commands: str | list[str] | None,
         resources=None,
         submit_dir=None,
         environment=None,
@@ -344,13 +343,13 @@ def submit(
             exit_code=returncode, stdout=stdout, stderr=stderr
         )
 
-    def get_job_info(self, job: Union[QJob, int, str]):
+    def get_job_info(self, job: QJob | int | str):
         pass
 
-    def get_jobs(self, jobs: List[Union[QJob, int, str]]):
+    def get_jobs(self, jobs: list[QJob | int | str]):
         pass
 
-    def cancel(self, job: Union[QJob, int, str]):
+    def cancel(self, job: QJob | int | str):
         cancel_cmd = self.get_cancel_cmd(job)
         stdout, stderr, returncode = self.execute_cmd(cancel_cmd)
         return self._parse_cancel_cmd_output(
@@ -363,5 +362,5 @@ def cancel(self, job: Union[QJob, int, str]):
     #     pass
 
     @abc.abstractmethod
-    def get_job(self, job: Union[QJob, int, str]):
+    def get_job(self, job: QJob | int | str):
         pass

src/qtoolkit/queue/slurm.py

@@ -1,6 +1,7 @@
+from __future__ import annotations
+
 import re
 from dataclasses import dataclass
-from typing import Union
 
 from qtoolkit.core.data_objects import (
     CancelResult,
@@ -212,9 +213,7 @@ def _parse_cancel_cmd_output(self, exit_code, stdout, stderr):
             status=status,
         )
 
-    def get_job(
-        self, job: Union[QJob, int, str], inplace=False, get_job_cmd="scontrol"
-    ):
+    def get_job(self, job: QJob | int | str, inplace=False, get_job_cmd="scontrol"):
         # TODO: there are two options to get info on a job in slurm:
         #  - scontrol show job JOB_ID
         #  - sacct -j JOB_ID