style: more typing

Signed-off-by: Yves Bastide <[email protected]>

Author: ybastide

simpleflow/cli/workflow.py

@@ -50,14 +50,14 @@ class OutputFormat(str, Enum):
 @app.command()
 def filter(
     ctx: typer.Context,
-    domain: Annotated[str, typer.Argument(envvar="SWF_DOMAIN")],
+    domain: Annotated[str, typer.Option(envvar="SWF_DOMAIN")],
     status: Annotated[Status, typer.Option("--status", "-s")] = Status.open,
     tag: str | None = None,
     workflow_id: str | None = None,
     workflow_type: str | None = None,
     workflow_type_version: str | None = None,
     close_status: CloseStatus | None = None,
-    started_since: int = 1,
+    started_since: Annotated[int, typer.Option("--started-since", "-n")] = 1,
     from_date: Annotated[datetime, typer.Option(formats=TIMESTAMP_FORMATS)] | None = None,
     to_date: Annotated[datetime, typer.Option(formats=TIMESTAMP_FORMATS)] | None = None,
 ):

simpleflow/process/_named_mixin.py

@@ -7,11 +7,9 @@
 from simpleflow import logger
 
 
-def with_state(state):
+def with_state(state: str):
     """
     Decorator used to change the process name when changing state.
-    :param state: new state
-    :type  state: str
     """
 
     def wrapper(method):
@@ -36,7 +34,7 @@ class NamedMixin:
         method explicitly if not the first parent)
     2- decorate your methods with "@with_state("my_state")"
 
-    You can optionnally expose some other attributes of your worker by defining
+    You can optionally expose some other attributes of your worker by defining
     the "_named_mixin_properties" attribute to a list or tuple of fields you want
     to include in your process title. For instance:
 
@@ -52,11 +50,11 @@ def __init__(self, *args, **kwargs):
         self.state = kwargs.get("state", "initializing")
 
     @property
-    def state(self):
+    def state(self) -> str:
         return self._state
 
     @state.setter
-    def state(self, value):
+    def state(self, value: str) -> None:
         self._state = value
         self.set_process_name()
 

simpleflow/process/_supervisor.py

@@ -86,7 +86,7 @@ def __init__(
         self._args = arguments if arguments is not None else ()
         self._background = background
 
-        self._processes = {}
+        self._processes: dict[int, psutil.Process] = {}
         self._terminating = False
 
         super().__init__()
@@ -106,7 +106,7 @@ def start(self):
 
     def _cleanup_worker_processes(self):
         # cleanup children
-        to_remove = []
+        to_remove: list[int] = []
         for pid, child in self._processes.items():
             try:
                 name, status = child.name(), child.status()

simpleflow/swf/process/decider/base.py

@@ -17,17 +17,17 @@
     from typing import Any
 
     from simpleflow.swf.executor import Executor
+    from simpleflow.swf.mapper.models.decision.base import Decision
     from simpleflow.swf.mapper.responses import Response
 
 
 class Decider(Supervisor):
     """
     Decider.
-
-    :ivar _poller: decider poller.
-    :type _poller: DeciderPoller
     """
 
+    _poller: DeciderPoller
+
     def __init__(self, poller, nb_children=None):
         self._poller = poller
         super().__init__(
@@ -74,15 +74,16 @@ def __init__(
         behind this is to limit operational burden by having a single service
         handling multiple workflows.
 
-        :param workflow_executors: executors handling workflow executions.
-        :type  workflow_executors: list[simpleflow.swf.executor.Executor]
+        workflow_executors: executors handling workflow executions.
 
         """
         self.workflow_name = f"{','.join([ex.workflow_class.name for ex in workflow_executors])}"
 
         # Maps a workflow's name to its definition.
         # Used to dispatch a decision task to the corresponding workflow.
-        self._workflow_executors = {executor.workflow_class.name: executor for executor in workflow_executors}
+        self._workflow_executors: dict[str, Executor] = {
+            executor.workflow_class.name: executor for executor in workflow_executors
+        }
 
         if task_list:
             self.task_list = task_list
@@ -123,8 +124,6 @@ def name(self):
         """
         The main purpose of this property is to find what workflow a decider
         handles.
-
-        :rtype: str
         """
         if self.workflow_name:
             suffix = f"(workflow={self.workflow_name})"
@@ -148,7 +147,6 @@ def complete(
         :param token: task token.
         :param decisions: decisions, maybe with context.
         :param execution_context: None...
-        :return: nothing.
         """
         if isinstance(decisions, DecisionsAndContext):
             decisions, execution_context = (
@@ -158,7 +156,7 @@ def complete(
         return simpleflow.swf.mapper.actors.Decider.complete(self, token, decisions, execution_context)
 
     @with_state("processing")
-    def process(self, decision_response):
+    def process(self, decision_response: Response) -> None:
         """
         Take a PollForDecisionTask response object and try to complete the
         decision task, by calling self._complete() with the response token and
@@ -171,7 +169,7 @@ def process(self, decision_response):
         spawn(self, decision_response)
 
     @with_state("deciding")
-    def decide(self, decision_response):
+    def decide(self, decision_response: Response) -> (list[Decision], DecisionsAndContext):
         """
         Delegate the decision to the decider worker (which itself delegates to
         the executor).