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 []