Merge pull request #48 from lonvia/enable-mypy-checks

Fix all warnings from mypy and flake8

.github/workflows/ci-build-tests.yml

@@ -4,22 +4,24 @@ jobs:
   build-tests:
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/checkout@v2
-    - name: Set up Python 3.9
-      uses: actions/setup-python@v2
+    - uses: actions/checkout@v4
+    - name: Set up Python 3.10
+      uses: actions/setup-python@v5
       with:
-        python-version: 3.9
+        python-version: "3.10"
     - name: Install python dependencies
       run: |
         python -m pip install --upgrade pip
-        pip install flake8 pytest pybind11
+        pip install flake8 pytest pybind11 mypy types-psycopg2 types-pyyaml
         pip install .
     - name: Lint with flake8
       run: |
         # stop the build if there are Python syntax errors or undefined names
         flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
         # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
-        flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
+        flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127  --exclude tests,build,__init__.py --extend-ignore=E302,E722 --statistics
+    - name: Check typing
+      run: mypy
     - name: Install PostgreSQL
       run: |
         sudo apt-get update

MANIFEST.in

@@ -1,3 +1,4 @@
 include src/nominatim_data_analyser/rules_specifications/*.yaml
 include src/nominatim_data_analyser/config/default.yaml
+include src/nominatim_data_analyser/py.typed
 recursive-include contrib *

cli.py

@@ -14,6 +14,6 @@
     sys.path.insert(0, str(SRC_DIR / BUILD_DIR))
 
 
-from nominatim_data_analyser.cli import cli
+from nominatim_data_analyser.cli import cli # noqa
 
 sys.exit(cli())

pyproject.toml

@@ -12,7 +12,7 @@ maintainers = [
 ]
 description = "QA Tool for Nominatim. Helps to improve the OpenStreetMap data quality and therefore the Nominatim search results."
 readme = "README.md"
-requires-python = ">=3.6"
+requires-python = ">=3.10"
 license = {text = "GPL-2.0-or-later"}
 classifiers = [
     "Programming Language :: Python :: 3",
@@ -46,7 +46,6 @@ packages = ["nominatim_data_analyser",
             "nominatim_data_analyser.core.pipes",
             "nominatim_data_analyser.core.pipes.rules_specific_pipes",
             "nominatim_data_analyser.core.pipes.rules_specific_pipes.place_nodes_close",
-            "nominatim_data_analyser.core.pipes.rules_specific_pipes.addresses_pipes",
             "nominatim_data_analyser.core.pipes.rules_specific_pipes.addr_house_number_no_digit",
             "nominatim_data_analyser.core.pipes.rules_specific_pipes.duplicate_label_role",
             "nominatim_data_analyser.core.pipes.rules_specific_pipes.same_wikidata",
@@ -65,3 +64,12 @@ nominatim-data-analyser = "nominatim_data_analyser.cli:cli"
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]
+
+[tool.mypy]
+strict = true
+python_version = "3.10"
+files = "src"
+
+[[tool.mypy.overrides]]
+module = ["geojson", "geojson.feature", "geojson.geometry"]
+ignore_missing_imports = true

src/nominatim_data_analyser/cli.py

@@ -1,21 +1,26 @@
 from .core.core import Core
 import argparse
 
-def cli():
+def cli() -> int:
     parser = argparse.ArgumentParser(prog='nominatim-analyser')
 
-    parser.add_argument('--execute-all', action='store_true', help='Executes all the QA rules')
-    parser.add_argument('--filter', metavar='<QA rules names>', nargs='+', help='Filters some QA rules so they are not executed.')
-    parser.add_argument('--execute-one', metavar='<QA rule name>', action='store', type=str, help='Executes the given QA rule')
+    parser.add_argument('--execute-all', action='store_true',
+                        help='Executes all the QA rules')
+    parser.add_argument('--filter', metavar='<QA rules names>', nargs='+',
+                        help='Filters some QA rules so they are not executed.')
+    parser.add_argument('--execute-one', metavar='<QA rule name>', action='store',
+                        type=str, help='Executes the given QA rule')
 
     args = parser.parse_args()
 
-    #Executes all the QA rules. If a filter is given, these rules are excluded from the execution.
+    # Executes all the QA rules. If a filter is given, these rules are excluded from the execution.
     if args.execute_all:
         if args.filter:
             Core().execute_all(args.filter)
         else:
             Core().execute_all()
     elif args.execute_one:
-        #Execute the given QA rule.
+        # Execute the given QA rule.
         Core().execute_one(args.execute_one)
+
+    return 0

src/nominatim_data_analyser/config/__init__.py

@@ -1 +1 @@
-from .config import Config
+from .config import Config as Config

src/nominatim_data_analyser/config/config.py

@@ -1,9 +1,10 @@
+from typing import Any
 from ..logger.logger import LOG
 from pathlib import Path
 import yaml
 
 class Config():
-    values: dict = dict()
+    values: dict[str, Any] = dict()
     default_config_folder_path = Path(__file__).parent
 
     @staticmethod

src/nominatim_data_analyser/core/__init__.py

@@ -1,4 +1,4 @@
 """
     Core module of the analyser
 """
-from .pipe import Pipe
+from .pipe import Pipe as Pipe

src/nominatim_data_analyser/core/assembler/__init__.py

@@ -2,5 +2,5 @@
     Module handling the assembly of a rule's pipeline.
 """
 
-from .pipeline_assembler import PipelineAssembler
-from .pipe_factory import PipeFactory
+from .pipeline_assembler import PipelineAssembler as PipelineAssembler
+from .pipe_factory import PipeFactory as PipeFactory

src/nominatim_data_analyser/core/assembler/pipe_factory.py

@@ -1,25 +1,23 @@
-from __future__ import annotations
+from typing import Any, cast
 from ..exceptions import YAMLSyntaxException
 from .. import pipes as pipes_module
-import typing
-
-if typing.TYPE_CHECKING: # pragma: no cover
-    from ..qa_rule import ExecutionContext
+from .. import Pipe
+from ..qa_rule import ExecutionContext
 
 class PipeFactory():
     """
         Factory to assemble pipes.
     """
     @staticmethod
-    def assemble_pipe(node_data: dict, exec_context: ExecutionContext):
+    def assemble_pipe(node_data: dict[str, Any], exec_context: ExecutionContext) -> Pipe:
         """
             Instantiate a pipe based on the given node_data
         """
         if 'type' not in node_data:
             raise YAMLSyntaxException("Each node of the tree (pipe) should have a type defined.")
 
         try:
-            assembled_pipe = getattr(pipes_module, node_data['type'])(node_data, exec_context)
+            assembled_pipe = cast(Pipe, getattr(pipes_module, node_data['type'])(node_data, exec_context))
         except AttributeError:
             raise YAMLSyntaxException(f"The type {node_data['type']} doesn't exist.")
 

src/nominatim_data_analyser/core/assembler/pipeline_assembler.py

@@ -1,41 +1,38 @@
-from __future__ import annotations
+from typing import Any
 from ..deconstructor import PipelineDeconstructor, BACKTRACKING_EVENT, NEW_NODE_EVENT
+from .. import Pipe
 from ..pipes import FillingPipe
 from ..qa_rule import ExecutionContext
 from ...logger.logger import LOG
 from collections import deque
 from .pipe_factory import PipeFactory
-from typing import Deque, Dict
-import typing
 
-if typing.TYPE_CHECKING: # pragma: no cover
-    from .. import Pipe
 
 class PipelineAssembler():
     """
-        Get deconstruction informations from the 
+        Get deconstruction informations from the
         pipeline deconstructor and assembles the final pipeline.
     """
-    def __init__(self, pipeline_specification: Dict, rule_name: str) -> None:
+    def __init__(self, pipeline_specification: dict[str, Any], rule_name: str) -> None:
         self.rule_name = rule_name
         self.deconstructor: PipelineDeconstructor = PipelineDeconstructor(pipeline_specification, rule_name)
         self.deconstructor.subscribe_event(NEW_NODE_EVENT, self.on_new_node)
         self.deconstructor.subscribe_event(BACKTRACKING_EVENT, self.on_backtrack)
-        self.first_pipe: Pipe = FillingPipe(None, None)
-        self.nodes_history: Deque[Pipe] = deque()
+        self.nodes_history: deque[Pipe] = deque()
         self.exec_context: ExecutionContext = ExecutionContext()
         self.exec_context.rule_name = rule_name
+        self.first_pipe: Pipe = FillingPipe({}, self.exec_context)
 
-    def on_new_node(self, node: dict) -> None:
+    def on_new_node(self, node: dict[str, Any]) -> None:
         """
             Raised by the deconstructor when a new node
             is reached.
         """
-        if node['type'] ==  'ROOT_NODE':
+        if node['type'] == 'ROOT_NODE':
             self.nodes_history.append(self.first_pipe)
         else:
             pipe = PipeFactory.assemble_pipe(node, self.exec_context)
-            #Plug the new pipe to the current last pipe of the deque
+            # Plug the new pipe to the current last pipe of the deque
             self.nodes_history[-1].plug_pipe(pipe)
             LOG.info("<%s> Assembler -> %s plugged to %s", self.rule_name, pipe, self.nodes_history[-1])
             self.nodes_history.append(pipe)

src/nominatim_data_analyser/core/core.py

@@ -4,7 +4,6 @@
 from ..logger.timer import Timer
 from ..config import Config
 from pathlib import Path
-from typing import Dict
 
 class Core():
     """
@@ -14,7 +13,7 @@ def __init__(self) -> None:
         Config.load_config()
         self.rules_path = Path(__file__, '..', '..', 'rules_specifications').resolve()
 
-    def execute_all(self, filter=None) -> None:
+    def execute_all(self, filter: list[str] | None = None) -> None:
         """
             Execute each QA rules.
 
@@ -30,6 +29,6 @@ def execute_one(self, name: str) -> None:
             Execute one QA rule based on its YAML file name.
         """
         timer = Timer().start_timer()
-        loaded_yaml: Dict = load_yaml_rule(name)
+        loaded_yaml = load_yaml_rule(name)
         PipelineAssembler(loaded_yaml, name).assemble().process_and_next()
         LOG.info('<%s> The whole rule executed in %s mins %s secs', name, *timer.get_elapsed())

src/nominatim_data_analyser/core/deconstructor/pipeline_deconstructor.py

@@ -1,4 +1,4 @@
-from typing import Callable, Deque, Dict, List
+from typing import Any, Callable
 from collections import deque
 from ...logger.logger import LOG
 
@@ -14,31 +14,32 @@ class PipelineDeconstructor():
         It uses a backtracking system to go back on upper nodes
         when reaching a leaf.
     """
-    def __init__(self, pipeline_specification: Dict, rule_name: str) -> None:
-        #Add a root node needed for the exploration of the tree
+    def __init__(self, pipeline_specification: dict[str, Any], rule_name: str) -> None:
+        # Add a root node needed for the exploration of the tree
         self._pipeline_specification = {
-            'ROOT_NODE': { 
+            'ROOT_NODE': {
                     'type': 'ROOT_NODE',
                     'out': pipeline_specification
                 }
         }
         self.rule_name = rule_name
-        self.current_node: Dict = None
-        self.nodes_history: Deque[Dict] = deque()
+        self.current_node: dict[str, Any] | None = None
+        self.nodes_history: deque[dict[str, Any]] = deque()
         self._init_event_callbacks()
 
     def deconstruct(self) -> None:
         """
             Explores the pipeline specification tree.
         """
-        self.current_node: Dict = self._pipeline_specification['ROOT_NODE']
+        self.current_node = self._pipeline_specification['ROOT_NODE']
         self._send_current_node_and_explore()
 
     def _send_current_node_and_explore(self) -> None:
         """
             Notifies that a new node has been reached and
             keep exploring the tree.
         """
+        assert self.current_node is not None
         self._notify_new_node(self.current_node)
         self._explore_deeper_or_backtrack()
 
@@ -47,40 +48,41 @@ def _explore_deeper_or_backtrack(self) -> None:
             If the current node still has an 'out' field, keep exploring
             deeper. Otherwise backtrack in the tree.
         """
+        assert self.current_node is not None
         if 'out' in self.current_node:
             self.nodes_history.append(self.current_node)
             self.current_node = self.current_node['out'].pop(next(iter(self.current_node['out'])))
             self._send_current_node_and_explore()
         else:
             self._backtrack()
-    
+
     def _backtrack(self) -> None:
         """
             Backtracks in the tree by getting the top node in the
-            nodes_history deque. 
+            nodes_history deque.
 
-            If there is no node left in the nodes_history, 
+            If there is no node left in the nodes_history,
             the exploring process is terminated naturally.
 
             Then keep exploring.
         """
         if self.nodes_history:
-            #backtrack
+            # backtrack
             self._notify_backtracking()
             self.current_node = self.nodes_history.popleft()
-            #Remove 'out' key if it became empty
+            # Remove 'out' key if it became empty
             if not self.current_node['out']:
                 self.current_node.pop('out', None)
             self._explore_deeper_or_backtrack()
 
-    def subscribe_event(self, event: str, callback: Callable) -> None:
+    def subscribe_event(self, event: str, callback: Callable[..., Any]) -> None:
         """
             Registers the given callback to the
             given event.
         """
         self._event_callbacks[event].append(callback)
-    
-    def _notify_new_node(self, node: dict) -> None:
+
+    def _notify_new_node(self, node: dict[str, Any]) -> None:
         """
             Notifies all subscribers that we reached a new node.
         """
@@ -94,10 +96,10 @@ def _notify_backtracking(self,) -> None:
         """
         LOG.info('<%s> Deconstruction -> BACKTRACK', self.rule_name)
         self._raise_event(BACKTRACKING_EVENT)
-    
-    def _raise_event(self, event_name: str, *args: any) -> None:
+
+    def _raise_event(self, event_name: str, *args: Any) -> None:
         """
-            Executes all registered callbacks of 
+            Executes all registered callbacks of
             the given event.
         """
         for callback in self._event_callbacks[event_name]:
@@ -107,6 +109,6 @@ def _init_event_callbacks(self) -> None:
         """
             Initializes all events and empty callbacks list.
         """
-        self._event_callbacks: Dict[str, List[Callable]] = dict()
+        self._event_callbacks: dict[str, list[Callable[..., Any]]] = dict()
         self._event_callbacks[NEW_NODE_EVENT] = []
         self._event_callbacks[BACKTRACKING_EVENT] = []

src/nominatim_data_analyser/core/dynamic_value/__init__.py

@@ -1,4 +1,5 @@
-from .dynamic_value import DynamicValue
-from .switch import Switch
-from .variable import Variable
-from .resolver import resolve_all, resolve_one
+from .dynamic_value import DynamicValue as DynamicValue
+from .switch import Switch as Switch
+from .variable import Variable as Variable
+from .resolver import (resolve_all as resolve_all,
+                       resolve_one as resolve_one)

src/nominatim_data_analyser/core/dynamic_value/dynamic_value.py

@@ -1,5 +1,5 @@
 from abc import ABCMeta, abstractmethod
-from typing import Any, Dict
+from typing import Any
 
 class DynamicValue(metaclass=ABCMeta):
     """
@@ -10,9 +10,9 @@ class DynamicValue(metaclass=ABCMeta):
         dynamically depending on the data values.
     """
     @abstractmethod
-    def resolve(self, data: Dict) -> Any:
+    def resolve(self, data: dict[str, Any]) -> Any:
         """
             Assigns a concrete value to the dynamic value
             based on the input data dictionnary.
         """
-        return
+        return