Add missing code on mypy errors

.prospector.yml

@@ -25,9 +25,8 @@ mypy:
     extra-checks: true
 
 pylint:
-  options:
-    extension-pkg-allow-list: mypy
   disable:
+    - c-extension-no-member # Needed for mypy
     - too-few-public-methods
     - missing-docstring
     - star-args

prospector/tools/mypy/__init__.py

@@ -1,22 +1,31 @@
+import json
 import re
 from multiprocessing import Process, Queue
-from typing import TYPE_CHECKING, Any, Callable, Optional
-
-from mypy import api
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Callable,
+    Optional,
+)
+
+import mypy.api
+import mypy.build
+import mypy.errors
+import mypy.fscache
+import mypy.main
 
 from prospector.finder import FileFinder
 from prospector.message import Location, Message
 from prospector.tools import ToolBase
-
-__all__ = ("MypyTool",)
-
 from prospector.tools.exceptions import BadToolConfig
 
 if TYPE_CHECKING:
     from prospector.config import ProspectorConfig
 
 _IGNORE_RE = re.compile(r"#\s*type:\s*ignore\[([^#]*[^# ])\](\s*#.*)?$", re.IGNORECASE)
 
+__all__ = ("MypyTool",)
+
 
 def format_message(message: str) -> Message:
     character: Optional[int]
@@ -61,9 +70,10 @@ def _run_in_subprocess(
 class MypyTool(ToolBase):
     def __init__(self, *args: Any, **kwargs: Any) -> None:
         super().__init__(*args, **kwargs)
-        self.checker = api
+        self.checker = mypy.api
         self.options = ["--show-column-numbers", "--no-error-summary"]
         self.use_dmypy = False
+        self.fscache = mypy.fscache.FileSystemCache()
 
     def configure(self, prospector_config: "ProspectorConfig", _: Any) -> None:
         options = prospector_config.tool_options("mypy")
@@ -93,21 +103,51 @@ def configure(self, prospector_config: "ProspectorConfig", _: Any) -> None:
             raise BadToolConfig("mypy", f"The option {name} has an unsupported value type: {type(value)}")
 
     def run(self, found_files: FileFinder) -> list[Message]:
-        paths = [str(path) for path in found_files.python_modules]
-        paths.extend(self.options)
+        args = [str(path) for path in found_files.python_modules]
+        args.extend(self.options)
         if self.use_dmypy:
             # Due to dmypy messing with stdout/stderr we call it in a separate
             # process
             q: Queue[str] = Queue(1)
-            p = Process(target=_run_in_subprocess, args=(q, self.checker.run_dmypy, ["run", "--"] + paths))
+            p = Process(target=_run_in_subprocess, args=(q, self.checker.run_dmypy, ["run", "--"] + args))
             p.start()
             result = q.get()
             p.join()
-        else:
-            result = self.checker.run(paths)
-        report, _ = result[0], result[1:]  # noqa
 
-        return [format_message(message) for message in report.splitlines()]
+            report, _ = result[0], result[1:]  # noqa
+            return [format_message(message) for message in report.splitlines()]
+        else:
+            return self._run_std(args)
+
+    def _run_std(self, args: list[str]) -> list[Message]:
+        sources, options = mypy.main.process_options(args, fscache=self.fscache)
+        options.output = "json"
+        res = mypy.build.build(sources, options, fscache=self.fscache)
+
+        messages = []
+        for mypy_json in res.errors:
+            mypy_message = json.loads(mypy_json)
+            message = f"{mypy_message['message']}."
+            if mypy_message.get("hint", ""):
+                message = f"{message}, Hint: {mypy_message['hint']}."
+            code = mypy_message["code"]
+            messages.append(
+                Message(
+                    "mypy",
+                    code=code,
+                    location=Location(
+                        path=mypy_message["file"],
+                        module=None,
+                        function=None,
+                        line=mypy_message["line"],
+                        character=mypy_message["column"],
+                    ),
+                    message=message,
+                    doc_url=f"{mypy.errors.BASE_RTD_URL}-{code}",
+                )
+            )
+
+        return messages
 
     def get_ignored_codes(self, line: str) -> list[str]:
         match = _IGNORE_RE.search(line)