Type annotations

Author: MetRonnie

cylc/flow/id.py

@@ -376,9 +376,7 @@ def submit_num(self) -> int | None:
             >>> Tokens('//c/t').submit_num is None
             True
         """
-        if self['job'] is None:
-            return None
-        return int(self['job'])
+        return int(self['job']) if self['job'] else None
 
     @overload
     def duplicate(

cylc/flow/rundb.py

@@ -715,12 +715,11 @@ def select_task_action_timers(self, callback):
             callback(row_idx, list(row))
 
     def select_task_job(
-        self, cycle: str, name: str, submit_num: Union[str, int, None] = None
-    ) -> Optional[Dict[str, Any]]:
+        self, cycle: str, name: str, submit_num: str | int | None = None
+    ) -> dict[str, Any] | None:
         """Select items from task_jobs by (cycle, name, submit_num).
 
         :return: a dict for mapping keys to the column values
-        :rtype: dict
         """
         keys = []
         for column in self.tables[self.TABLE_TASK_JOBS].columns[3:]:
@@ -739,7 +738,7 @@ def select_task_job(
             '''  # nosec B608
             # * table name is code constant
             # * keys are code constants
-            stmt_args: List[Any] = [cycle, name]
+            stmt_args: list[Any] = [cycle, name]
         else:
             stmt = rf'''
                 SELECT

cylc/flow/task_events_mgr.py

@@ -221,9 +221,9 @@ def get_event_id(event: str, itask: 'TaskProxy') -> str:
 def log_task_job_activity(
     ctx: 'SubProcContext',
     workflow: str,
-    point: Union[str, 'PointBase'],
+    point: 'str | PointBase',
     name: str,
-    submit_num: Union[str, int, None] = None,
+    submit_num: str | int | None = None,
 ):
     """Log an activity for a task job."""
     ctx_str = str(ctx)
@@ -766,7 +766,7 @@ def process_message(
         if job_aborted:
             task_output = TASK_OUTPUT_FAILED
 
-        output_completed: Optional[bool] = False
+        output_completed: bool | None = False
         if task_output not in {TASK_OUTPUT_SUBMIT_FAILED, TASK_OUTPUT_FAILED}:
             output_completed = (
                 itask.state.outputs.set_message_complete(task_output, forced)
@@ -1372,7 +1372,7 @@ def _process_message_failed(
         message: str,
         forced: bool,
         full_message: str,
-        run_signal: Optional[str] = None,
+        run_signal: str | None = None,
     ) -> bool:
         """Helper for process_message, handle a failed message.
 
@@ -1421,7 +1421,7 @@ def _process_job_failed(
         self,
         itask: 'TaskProxy',
         event_time: str,
-        run_signal: Optional[str] = None,
+        run_signal: str | None = None,
     ):
         itask.set_summary_time('finished', event_time)
         self.data_store_mgr.delta_job_time(itask, 'finished', event_time)

cylc/flow/task_job_logs.py

@@ -16,11 +16,7 @@
 """Define job log filenames and option names."""
 
 import os
-from typing import (
-    TYPE_CHECKING,
-    Optional,
-    Union,
-)
+from typing import TYPE_CHECKING
 
 from cylc.flow.id import Tokens
 from cylc.flow.pathutil import get_workflow_run_job_dir
@@ -52,10 +48,10 @@
 
 def get_task_job_log(
     workflow: str,
-    point: Union[str, 'PointBase'],
+    point: 'str | PointBase',
     name: str,
-    submit_num: Union[str, int, None] = None,
-    suffix: Optional[str] = None,
+    submit_num: str | int | None = None,
+    suffix: str | None = None,
 ):
     """Return the full job log path."""
     args = [
@@ -73,9 +69,9 @@ def get_task_job_log(
 
 def get_task_job_activity_log(
     workflow: str,
-    point: Union[str, 'PointBase'],
+    point: 'str | PointBase',
     name: str,
-    submit_num: Union[str, int, None] = None,
+    submit_num: str | int | None = None,
 ):
     """Shorthand for get_task_job_log(..., suffix="job-activity.log")."""
     return get_task_job_log(
@@ -85,9 +81,9 @@ def get_task_job_activity_log(
 
 def get_task_job_job_log(
     workflow: str,
-    point: Union[str, 'PointBase'],
+    point: 'str | PointBase',
     name: str,
-    submit_num: Union[str, int, None] = None,
+    submit_num: str | int | None = None,
 ):
     """Shorthand for get_task_job_log(..., suffix="job")."""
     return get_task_job_log(workflow, point, name, submit_num, JOB_LOG_JOB)

cylc/flow/task_message.py

@@ -29,11 +29,7 @@
 )
 import os
 import sys
-from typing import (
-    List,
-    Optional,
-    Tuple,
-)
+from typing import List
 
 from cylc.flow.exceptions import WorkflowStopped
 import cylc.flow.flags
@@ -82,7 +78,7 @@
 '''
 
 
-def split_run_signal(message: str) -> Tuple[str, Optional[str]]:
+def split_run_signal(message: str) -> tuple[str, str | None]:
     """Get the run signal from a message.
 
     >>> split_run_signal('failed/ERR')