mass reformat & upgrade to Python 3.10+ syntax

Author: mr-c

contrib/mypy-stubs/pubsub/core/topictreetraverser.pyi

@@ -1,7 +1,8 @@
-from .topicobj import Topic as Topic
 from enum import IntEnum
 from typing import Optional
 
+from .topicobj import Topic as Topic
+
 class ITopicTreeVisitor: ...
 
 class TreeTraversal(IntEnum):
@@ -12,4 +13,6 @@ class TreeTraversal(IntEnum):
 class TopicTreeTraverser:
     def __init__(self, visitor: Optional[ITopicTreeVisitor] = None) -> None: ...
     def setVisitor(self, visitor: ITopicTreeVisitor) -> None: ...
-    def traverse(self, topicObj: Topic, how: TreeTraversal = ..., onlyFiltered: bool = True) -> None: ...
+    def traverse(
+        self, topicObj: Topic, how: TreeTraversal = ..., onlyFiltered: bool = True
+    ) -> None: ...

src/toil/__init__.py

@@ -16,14 +16,11 @@
 import re
 import socket
 import sys
-from datetime import datetime
-from typing import TYPE_CHECKING, Optional
-
-from toil.lib.web import web_session
+from typing import TYPE_CHECKING
 
 from docker.errors import ImageNotFound
 from toil.lib.memoize import memoize
-from toil.lib.retry import retry as retry
+from toil.lib.web import web_session
 from toil.version import currentCommit
 
 if TYPE_CHECKING:
@@ -32,7 +29,7 @@
 log = logging.getLogger(__name__)
 
 
-def which(cmd, mode=os.F_OK | os.X_OK, path=None) -> Optional[str]:
+def which(cmd, mode=os.F_OK | os.X_OK, path=None) -> str | None:
     """
     Return the path with conforms to the given mode on the Path.
 
@@ -479,4 +476,3 @@ def logProcessContext(config: "Config") -> None:
 
     log.info("Running Toil version %s on host %s.", version, socket.gethostname())
     log.debug("Configuration: %s", config.__dict__)
-

src/toil/batchSystems/abstractBatchSystem.py

@@ -11,7 +11,6 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-from dataclasses import dataclass
 import enum
 import logging
 import os
@@ -21,8 +20,9 @@
 from argparse import ArgumentParser, _ArgumentGroup
 from collections.abc import Iterator
 from contextlib import contextmanager
+from dataclasses import dataclass
 from threading import Condition
-from typing import Any, ContextManager, NamedTuple, Optional, Union, cast
+from typing import Any, ContextManager, NamedTuple, cast
 
 from toil.batchSystems.options import OptionSetter
 from toil.bus import MessageBus, MessageOutbox
@@ -73,6 +73,7 @@ def to_string(cls, value: int) -> str:
         except ValueError:
             return str(value)
 
+
 @dataclass
 class UpdatedBatchJobInfo:
     jobID: int
@@ -87,20 +88,20 @@ class UpdatedBatchJobInfo:
     (e.g. job is lost, or otherwise died but actual exit code was not reported).
     """
 
-    exitReason: Optional[BatchJobExitReason] = None
-    wallTime: Union[float, int, None] = None
-    backing_id: Optional[str] = None
+    exitReason: BatchJobExitReason | None = None
+    wallTime: float | int | None = None
+    backing_id: str | None = None
     """
     The identifier for the job in the backing scheduler, if available.
     """
 
 
 # Information required for worker cleanup on shutdown of the batch system.
 class WorkerCleanupInfo(NamedTuple):
-    work_dir: Optional[str]
+    work_dir: str | None
     """Work directory path (where the cache would go) if specified by user"""
 
-    coordination_dir: Optional[str]
+    coordination_dir: str | None
     """Coordination directory path (where lock files would go) if specified by user"""
 
     workflow_id: str
@@ -171,7 +172,7 @@ def issueBatchJob(
         self,
         command: str,
         job_desc: JobDescription,
-        job_environment: Optional[dict[str, str]] = None,
+        job_environment: dict[str, str] | None = None,
     ) -> int:
         """
         Issues a job with the specified command to the batch system and returns
@@ -224,7 +225,7 @@ def getRunningBatchJobIDs(self) -> dict[int, float]:
         raise NotImplementedError()
 
     @abstractmethod
-    def getUpdatedBatchJob(self, maxWait: int) -> Optional[UpdatedBatchJobInfo]:
+    def getUpdatedBatchJob(self, maxWait: int) -> UpdatedBatchJobInfo | None:
         """
         Returns information about job that has updated its status (i.e. ceased
         running, either successfully or with an error). Each such job will be
@@ -242,7 +243,7 @@ def getUpdatedBatchJob(self, maxWait: int) -> Optional[UpdatedBatchJobInfo]:
         """
         raise NotImplementedError()
 
-    def getSchedulingStatusMessage(self) -> Optional[str]:
+    def getSchedulingStatusMessage(self) -> str | None:
         """
         Get a log message fragment for the user about anything that might be
         going wrong in the batch system, if available.
@@ -269,7 +270,7 @@ def shutdown(self) -> None:
         """
         raise NotImplementedError()
 
-    def setEnv(self, name: str, value: Optional[str] = None) -> None:
+    def setEnv(self, name: str, value: str | None = None) -> None:
         """
         Set an environment variable for the worker process before it is launched.
 
@@ -286,7 +287,7 @@ def setEnv(self, name: str, value: Optional[str] = None) -> None:
         raise NotImplementedError()
 
     @classmethod
-    def add_options(cls, parser: Union[ArgumentParser, _ArgumentGroup]) -> None:
+    def add_options(cls, parser: ArgumentParser | _ArgumentGroup) -> None:
         """
         If this batch system provides any command line options, add them to the given parser.
         """
@@ -351,7 +352,7 @@ def __init__(
                 workflow_id=config.workflowID,
                 clean_work_dir=config.cleanWorkDir,
             )
-        self._outbox: Optional[MessageOutbox] = None
+        self._outbox: MessageOutbox | None = None
 
     def check_resource_request(self, requirer: Requirer) -> None:
         """
@@ -400,7 +401,7 @@ def _check_accelerator_request(self, requirer: Requirer) -> None:
                 details=["The batch system does not support any accelerators."],
             )
 
-    def setEnv(self, name: str, value: Optional[str] = None) -> None:
+    def setEnv(self, name: str, value: str | None = None) -> None:
         """
         Set an environment variable for the worker process before it is launched. The worker
         process will typically inherit the environment of the machine it is running on but this
@@ -572,7 +573,7 @@ class AbstractScalableBatchSystem(AbstractBatchSystem):
 
     @abstractmethod
     def getNodes(
-        self, preemptible: Optional[bool] = None, timeout: int = 600
+        self, preemptible: bool | None = None, timeout: int = 600
     ) -> dict[str, NodeInfo]:
         """
         Returns a dictionary mapping node identifiers of preemptible or non-preemptible nodes to
@@ -622,9 +623,9 @@ def __init__(
         self,
         requirer: Requirer,
         resource: str,
-        available: Optional[ParsedRequirement] = None,
-        batch_system: Optional[str] = None,
-        source: Optional[str] = None,
+        available: ParsedRequirement | None = None,
+        batch_system: str | None = None,
+        source: str | None = None,
         details: list[str] = [],
     ) -> None:
         """
@@ -639,7 +640,7 @@ def __init__(
         :param details: Any extra details about the problem that can be attached to the error.
         """
 
-        self.job_name: Optional[str] = str(requirer)
+        self.job_name: str | None = str(requirer)
         self.resource = resource
         self.requested = cast(ParsedRequirement, getattr(requirer, resource))
         self.available = available
@@ -695,8 +696,8 @@ class AcquisitionTimeoutException(Exception):
     def __init__(
         self,
         resource: str,
-        requested: Union[int, float, set[int]],
-        available: Union[int, float, set[int]],
+        requested: int | float | set[int],
+        available: int | float | set[int],
     ) -> None:
         """
         Creates an instance of this exception that indicates which resource is insufficient for

src/toil/batchSystems/abstractGridEngineBatchSystem.py

@@ -17,7 +17,6 @@
 from datetime import datetime
 from queue import Empty, Queue
 from threading import Lock, Thread
-from typing import Optional, Union
 
 from toil.batchSystems.abstractBatchSystem import (
     BatchJobExitReason,
@@ -27,9 +26,9 @@
 from toil.bus import ExternalBatchIdMessage, get_job_kind
 from toil.common import Config
 from toil.job import AcceleratorRequirement, JobDescription
-from toil.statsAndLogging import TRACE
 from toil.lib.misc import CalledProcessErrorStderr
 from toil.lib.retry import DEFAULT_DELAYS, old_retry
+from toil.statsAndLogging import TRACE
 
 logger = logging.getLogger(__name__)
 
@@ -172,7 +171,7 @@ def createJobs(self, newJob: JobTuple) -> bool:
                     "Job %s with batch system ID %s queued as job %s",
                     jobName,
                     jobID,
-                    str(batchJobID)
+                    str(batchJobID),
                 )
 
                 # Store dict for mapping Toil job ID to batch job ID
@@ -258,8 +257,12 @@ def checkOnJobs(self):
                     self.coalesce_job_exit_codes, batch_job_id_list
                 )
                 # We got the statuses as a batch
-                for running_job_id, status, backing_id in zip(running_job_list, statuses, batch_job_id_list):
-                    activity = self._handle_job_status(running_job_id, status, activity, backing_id)
+                for running_job_id, status, backing_id in zip(
+                    running_job_list, statuses, batch_job_id_list
+                ):
+                    activity = self._handle_job_status(
+                        running_job_id, status, activity, backing_id
+                    )
 
             self._checkOnJobsCache = activity
             self._checkOnJobsTimestamp = datetime.now()
@@ -268,7 +271,7 @@ def checkOnJobs(self):
         def _handle_job_status(
             self,
             job_id: int,
-            status: Union[int, tuple[int, Optional[BatchJobExitReason]], None],
+            status: int | tuple[int, BatchJobExitReason | None] | None,
             activity: bool,
             backing_id: str,
         ) -> bool:
@@ -313,7 +316,9 @@ def _runStep(self):
             if self.checkOnJobs():
                 activity = True
             if not activity:
-                logger.log(TRACE, "No activity, sleeping for %is", self.boss.sleepSeconds())
+                logger.log(
+                    TRACE, "No activity, sleeping for %is", self.boss.sleepSeconds()
+                )
             return True
 
         def run(self):
@@ -332,7 +337,7 @@ def run(self):
 
         def coalesce_job_exit_codes(
             self, batch_job_id_list: list
-        ) -> list[Union[int, tuple[int, Optional[BatchJobExitReason]], None]]:
+        ) -> list[int | tuple[int, BatchJobExitReason | None] | None]:
             """
             Returns exit codes and possibly exit reasons for a list of jobs, or None if they are running.
 
@@ -362,8 +367,8 @@ def prepareSubmission(
             jobID: int,
             command: str,
             jobName: str,
-            job_environment: Optional[dict[str, str]] = None,
-            gpus: Optional[int] = None,
+            job_environment: dict[str, str] | None = None,
+            gpus: int | None = None,
         ) -> list[str]:
             """
             Preparation in putting together a command-line string
@@ -416,7 +421,7 @@ def killJob(self, jobID):
         @abstractmethod
         def getJobExitCode(
             self, batchJobID
-        ) -> Union[int, tuple[int, Optional[BatchJobExitReason]], None]:
+        ) -> int | tuple[int, BatchJobExitReason | None] | None:
             """
             Returns job exit code and possibly an instance of abstractBatchSystem.BatchJobExitReason.
 
@@ -478,11 +483,10 @@ def issueBatchJob(
         self,
         command: str,
         job_desc: JobDescription,
-        job_environment: Optional[dict[str, str]] = None,
+        job_environment: dict[str, str] | None = None,
     ):
         # Avoid submitting internal jobs to the batch queue, handle locally
-        local_id = self.handleLocalJob(command, job_desc)
-        if local_id is not None:
+        if (local_id := self.handleLocalJob(command, job_desc)) is not None:
             return local_id
         else:
             self.check_resource_request(job_desc)

src/toil/batchSystems/awsBatch.py

@@ -35,7 +35,7 @@
 import uuid
 from argparse import ArgumentParser, _ArgumentGroup
 from collections.abc import Iterator
-from typing import Any, Optional, Union
+from typing import Any
 
 from botocore.exceptions import ClientError
 
@@ -137,15 +137,15 @@ def __init__(
         # is managed by the BatchSystemLocalSupport.
 
         # Here is where we will store the user script resource object if we get one.
-        self.user_script: Optional[Resource] = None
+        self.user_script: Resource | None = None
 
         # Get the image to deploy from Toil's configuration
         self.docker_image = applianceSelf()
 
         # We can't use AWS Batch without a job definition. But we can use one
         # of them for all the jobs. We want to lazily initialize it. This will
         # be an ARN.
-        self.job_definition: Optional[str] = None
+        self.job_definition: str | None = None
 
         # We need a way to map between our batch system ID numbers, and AWS Batch job IDs from the server.
         self.bs_id_to_aws_id: dict[int, str] = {}
@@ -179,11 +179,10 @@ def issueBatchJob(
         self,
         command: str,
         job_desc: JobDescription,
-        job_environment: Optional[dict[str, str]] = None,
+        job_environment: dict[str, str] | None = None,
     ) -> int:
         # Try the job as local
-        local_id = self.handleLocalJob(command, job_desc)
-        if local_id is not None:
+        if (local_id := self.handleLocalJob(command, job_desc)) is not None:
             # It is a local job
             return local_id
         else:
@@ -298,7 +297,7 @@ def _ensafen_name(input_name: str) -> str:
         # And re-compose them into a string
         return "".join(kept_chars)
 
-    def _get_runtime(self, job_detail: dict[str, Any]) -> Optional[float]:
+    def _get_runtime(self, job_detail: dict[str, Any]) -> float | None:
         """
         Internal function. Should not be called outside this class.
 
@@ -349,7 +348,7 @@ def _get_exit_code(self, job_detail: dict[str, Any]) -> int:
             )
         )
 
-    def getUpdatedBatchJob(self, maxWait: int) -> Optional[UpdatedBatchJobInfo]:
+    def getUpdatedBatchJob(self, maxWait: int) -> UpdatedBatchJobInfo | None:
         # Remember when we started, for respecting the timeout
         entry = datetime.datetime.now()
         while (
@@ -494,7 +493,7 @@ def _get_or_create_job_definition(self) -> str:
         if self.job_definition is None:
             # First work out what volume mounts to make, because the type
             # system is happiest this way
-            volumes: list[dict[str, Union[str, dict[str, str]]]] = []
+            volumes: list[dict[str, str | dict[str, str]]] = []
             mount_points: list[dict[str, str]] = []
             for i, shared_path in enumerate(
                 {
@@ -650,7 +649,7 @@ def killBatchJobs(self, job_ids: list[int]) -> None:
                 self._wait_until_stopped(self.bs_id_to_aws_id[bs_id])
 
     @classmethod
-    def add_options(cls, parser: Union[ArgumentParser, _ArgumentGroup]) -> None:
+    def add_options(cls, parser: ArgumentParser | _ArgumentGroup) -> None:
         parser.add_argument(
             "--awsBatchRegion",
             dest="aws_batch_region",