Source code for prospector.tools.mypy
import json
import re
from multiprocessing import Process, Queue
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
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]
try:
(path, line_str, char_str, err_type, err_msg) = message.split(":", 4)
line = int(line_str)
character = int(char_str)
except ValueError:
try:
(path, line_str, err_type, err_msg) = message.split(":", 3)
line = int(line_str)
character = None
except ValueError:
(path, err_type, err_msg) = message.split(":", 2)
line = 0
character = None
location = Location(
path=path,
module=None,
function=None,
line=line,
character=character,
)
return Message(
source="mypy",
code=err_type.lstrip(" "),
location=location,
message=err_msg.lstrip(" "),
)
def _run_in_subprocess(
q: "Queue[tuple[str, str]]", cmd: Callable[[list[str]], tuple[str, str]], paths: list[str]
) -> None:
"""
This function exists only to be called by multiprocessing.Process as using
lambda is forbidden
"""
q.put(cmd(paths))
[docs]
class MypyTool(ToolBase):
def __init__(self, *args: Any, **kwargs: Any) -> None:
super().__init__(*args, **kwargs)
self.checker = mypy.api
self.options = ["--show-column-numbers", "--no-error-summary"]
self.use_dmypy = False
self.fscache = mypy.fscache.FileSystemCache()
[docs]
def configure(self, prospector_config: "ProspectorConfig", _: Any) -> None:
options = prospector_config.tool_options("mypy")
self.use_dmypy = options.pop("use-dmypy", False)
# For backward compatibility
if "follow-imports" not in options:
options["follow-imports"] = "normal"
if "python-2-mode" in options and "py2" not in options:
options["py2"] = options.pop("python-2-mode")
for name, value in options.items():
if value is False:
continue
if value is True:
self.options.append(f"--{name}")
continue
if isinstance(value, (int, float, str)):
self.options.append(f"--{name}={value}")
continue
if isinstance(value, list):
for v in value:
self.options.append(f"--{name}-{v}")
continue
raise BadToolConfig("mypy", f"The option {name} has an unsupported value type: {type(value)}")
for code in prospector_config.get_disabled_messages("mypy"):
self.options.append(f"--disable-error-code={code}")
for code in prospector_config.get_enabled_messages("mypy"):
self.options.append(f"--enable-error-code={code}")
[docs]
def run(self, found_files: FileFinder) -> list[Message]:
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", "--"] + args))
p.start()
result = q.get()
p.join()
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]:
messages = []
try:
sources, options = mypy.main.process_options(args, fscache=self.fscache)
except (SystemExit, Exception) as e:
message = "The error(s) will be displayed before the messages" if isinstance(e, SystemExit) else str(e)
messages.append(
Message(
"mypy",
code="fatal-options-error",
message=message,
location=Location(
path="",
module=None,
function=None,
line=0,
character=0,
),
)
)
return messages
options.output = "json"
try:
res = mypy.build.build(sources, options, fscache=self.fscache)
except Exception as e:
messages.append(
Message(
"mypy",
code="fatal-build-error",
message=str(e),
location=Location(
path="",
module=None,
function=None,
line=0,
character=0,
),
)
)
return 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} {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
[docs]
def get_ignored_codes(self, line: str) -> list[tuple[str, int]]:
match = _IGNORE_RE.search(line)
if match:
return [(e.strip(), 0) for e in match.group(1).split(",")]
return []