Add type hints and make mypy compatible (#192)

* add type hints
enable mypy check
enable test and linting check on pull request
drop python2 support

Closes #189

* split push and pull request

* fix typo in mypy file

* Adding to the typehints

* Remove old py2 code

* Add mypy to pre-commit and github pipeline

* Improving the type hints of the Node decorator

* fix mypy

* fix python3.9 issues

* fix python import issues

* Update .github/workflows/pytest.yml

Co-authored-by: Grische <[email protected]>

* mypy python version matrix

* only explicit versios

* fix mypy version matrix

---------

Co-authored-by: Paul Schweizer <[email protected]>
Co-authored-by: Grische <[email protected]>

Author: 3 people

.github/workflows/mypy.yml

@@ -0,0 +1,38 @@
+name: Mypy
+
+on:
+  push:
+    branches:
+      - "master"
+      - "main"
+  pull_request:
+    branches:
+      - "master"
+      - "main"
+  workflow_dispatch:
+
+jobs:
+  mypy:
+    name: Mypy
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.9", "3.10", "3.11", "3.x"]
+    steps:
+      - name: Check out repository code
+        uses: actions/checkout@v5
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v6
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Install Poetry
+        run: |
+          python -m pip install --upgrade poetry wheel
+
+      - name: Install dependencies
+        run: |
+          poetry install
+
+      - name: Run mypy type checks
+        run: |
+          poetry run mypy ./flowpipe

.pre-commit-config.yaml

@@ -7,13 +7,18 @@ repos:
     rev: 5.12.0
     hooks:
       - id: isort
-  - repo: https://github.com/python-poetry/poetry
-    rev: 1.7.0
+  - repo: local
     hooks:
       - id: poetry-check
+        name: poetry-check
+        entry: poetry check
+        language: system
+        pass_filenames: false
       - id: poetry-lock
-  - repo: local
-    hooks:
+        name: poetry-lock
+        entry: poetry lock
+        language: system
+        pass_filenames: false
       - id: unittests
         name: unittests
         language: system
@@ -24,3 +29,9 @@ repos:
         language: system
         entry: poetry run pylint ./flowpipe
         pass_filenames: false
+      - id: mypy
+        name: mypy (flowpipe)
+        entry: poetry run mypy flowpipe
+        language: system
+        pass_filenames: false
+        files: ^flowpipe/.*\.py$

flowpipe/evaluator.py

@@ -1,21 +1,28 @@
 """Classes to evaluate flowpipe Graphs in various ways."""
 
+from __future__ import annotations
+
 import logging
 import time
 from concurrent import futures
 from multiprocessing import Manager, Process
 from pickle import PicklingError
+from typing import TYPE_CHECKING
 
 from .errors import FlowpipeMultiprocessingError
 
+if TYPE_CHECKING:  # pragma: no cover
+    from .graph import Graph
+    from .node import INode
+
 log = logging.getLogger(__name__)
 
 
 class Evaluator:
     """An engine to evaluate a Graph."""
 
     @staticmethod
-    def _evaluation_sequence(graph):
+    def _evaluation_sequence(graph: Graph) -> list[INode]:
         """Sort Nodes into a sequential, flat execution order.
 
         Replicated here for flexibility; defaults to Graph's implementation.
@@ -27,18 +34,20 @@ def _evaluation_sequence(graph):
         """
         return graph.evaluation_sequence
 
-    def _nodes_to_evaluate(self, graph, skip_clean):
+    def _nodes_to_evaluate(
+        self, graph: Graph, skip_clean: bool
+    ) -> list[INode]:
         """Get the nodes to evaluate, in order."""
         nodes = self._evaluation_sequence(graph)
         if skip_clean:
             nodes = [n for n in nodes if n.is_dirty]
         return nodes
 
-    def _evaluate_nodes(self, nodes):
+    def _evaluate_nodes(self, nodes: list[INode]) -> None:
         """Perform the actual node evaluation."""
         raise NotImplementedError  # pragma: no cover
 
-    def evaluate(self, graph, skip_clean=False):
+    def evaluate(self, graph: Graph, skip_clean: bool = False) -> None:
         """Evaluate the graph.
 
         Args:
@@ -55,7 +64,7 @@ def evaluate(self, graph, skip_clean=False):
 class LinearEvaluator(Evaluator):
     """Evaluate the graph linearly in a single thread."""
 
-    def _evaluate_nodes(self, nodes):
+    def _evaluate_nodes(self, nodes: list[INode]) -> None:
         """Evaluate the graph linearly in a single thread.
 
         Args:
@@ -69,18 +78,17 @@ def _evaluate_nodes(self, nodes):
 class ThreadedEvaluator(Evaluator):
     """Evaluate each node in a separate thread."""
 
-    def __init__(self, max_workers=None):
+    def __init__(self, max_workers: int | None = None):
         """Intialize with the graph and how many threads to use.
 
         Args:
-            graph (flowpipe.Graph): The graph to evaluate.
             max_workers (int): The number of threads to use in parallel,
                 defaults to the futures.ThreadPoolExecutor default.
 
         """
         self.max_workers = max_workers
 
-    def _evaluate_nodes(self, nodes):
+    def _evaluate_nodes(self, nodes: list[INode]) -> None:
         """Evaluate each node in a separate thread.
 
         Args:
@@ -90,11 +98,11 @@ def _evaluate_nodes(self, nodes):
         # create copy to prevent side effects
         nodes_to_evaluate = list(nodes)
 
-        def node_runner(node):
+        def node_runner(node: INode) -> INode:
             node.evaluate()
             return node
 
-        running_futures = {}
+        running_futures: dict = {}
         with futures.ThreadPoolExecutor(max_workers=self.max_workers) as tpe:
             while nodes_to_evaluate or running_futures:
                 log.debug(
@@ -148,7 +156,7 @@ def node_runner(node):
 class LegacyMultiprocessingEvaluator(Evaluator):
     """Evaluate nodes in separate processes."""
 
-    def __init__(self, submission_delay=0.1):
+    def __init__(self, submission_delay: float = 0.1):
         """Initialize with the graph and the delay between launching nodes.
 
         Args:
@@ -158,14 +166,14 @@ def __init__(self, submission_delay=0.1):
         """
         self.submission_delay = submission_delay
 
-    def _evaluate_nodes(self, nodes):
+    def _evaluate_nodes(self, nodes: list[INode]):
         # create copy to prevent side effects
         nodes_to_evaluate = list(nodes)
         manager = Manager()
         nodes_data = manager.dict()
-        processes = {}
+        processes: dict[str, Process] = {}
 
-        def upstream_ready(node):
+        def upstream_ready(node: INode) -> bool:
             """Check whether all upstream nodes have been evaluated."""
             for upstream in node.upstream_nodes:
                 if upstream in nodes_to_evaluate:
@@ -208,7 +216,7 @@ def upstream_ready(node):
             time.sleep(self.submission_delay)
 
 
-def _evaluate_node_in_process(identifier, nodes_data):
+def _evaluate_node_in_process(identifier: str, nodes_data: dict):
     """Evaluate a node when multiprocessing.
 
     1. Deserializing the node from the given nodes_data dict
@@ -253,7 +261,7 @@ def _evaluate_node_in_process(identifier, nodes_data):
     nodes_data[identifier] = data
 
 
-def _update_node(node, data):
+def _update_node(node: INode, data: dict):
     """Apply the plug values of the data dict to the node object."""
     for name, input_plug in data["inputs"].items():
         node.inputs[name].value = input_plug["value"]

flowpipe/event.py

@@ -2,7 +2,9 @@
 
 They an be used to observe the evaluation process.
 """
+
 import logging
+from collections.abc import Callable
 
 log = logging.getLogger(__name__)
 
@@ -14,38 +16,38 @@ class Event:
     Please note that the integrity of the listeners is not enforced or checked.
     """
 
-    def __init__(self, name):
+    def __init__(self, name: str):
         """Initialize the list of listeners
 
         Args:
             name (str): The (unique) name of the signal
         """
         self.name = name
-        self._listeners = []
+        self._listeners:list[Callable] = []
 
-    def emit(self, *args, **kwargs):
+    def emit(self, *args, **kwargs) -> None:
         """Call all the listeners with the given args and kwargs."""
         for listener in self._listeners:
             listener(*args, **kwargs)
 
-    def register(self, listener):
+    def register(self, listener: Callable) -> None:
         """Register the given function object if it is not yet registered."""
         if not self.is_registered(listener):
             self._listeners.append(listener)
 
-    def deregister(self, listener):
+    def deregister(self, listener: Callable) -> None:
         """Deregister the given function object if it is registered."""
         if self.is_registered(listener):
             self._listeners.pop(self._listeners.index(listener))
             log.debug("%s deregistered", listener)
         else:
             log.exception("%s was never registered", listener)
 
-    def is_registered(self, listener):
+    def is_registered(self, listener: Callable) -> bool:
         """Whether the given function object is already registered."""
         return listener in self._listeners
 
-    def clear(self):
+    def clear(self) -> None:
         """Remove all listeners from this event."""
         for listener in self._listeners:
             self.deregister(listener)