Allow setting mypy/dmypy commands via config

README.rst

@@ -78,6 +78,16 @@ Configuration
      - ``normal``, ``silent``, ``skip`` or ``error``
      - ``mypy`` **parameter** ``follow-imports``. In ``mypy`` this is ``normal`` by default. We set it ``silent``, to sort out unwanted results. This can cause cash invalidation if you also run ``mypy`` in other ways. Setting this to ``normal`` avoids this at the cost of a small performance penalty.
      - ``silent``
+   * - ``mypy_command``
+     - ``pylsp.plugins.pylsp_mypy.mypy_command``
+     - ``array`` of ``string`` items
+     - **The command to run mypy**. This is useful if you want to run mypy in a specific virtual environment.
+     - ``[]``
+   * - ``dmypy_command``
+     - ``pylsp.plugins.pylsp_mypy.dmypy_command``
+     - ``array`` of ``string`` items
+     - **The command to run dmypy**. This is useful if you want to run dmypy in a specific virtual environment.
+     - ``[]``
 
 Using a ``pyproject.toml`` for configuration, which is in fact the preferred way, your configuration could look like this:
 
@@ -151,6 +161,26 @@ With ``report_progress`` your config could look like this:
         "report_progress": True
     }
 
+With ``mypy_command`` your config could look like this:
+
+::
+
+    {
+        "enabled": True,
+        "mypy_command": ["poetry", "run", "mypy"]
+    }
+
+With ``dmypy_command`` your config could look like this:
+
+::
+
+    {
+        "enabled": True,
+        "live_mode": False,
+        "dmypy": True,
+        "dmypy_command": ["/path/to/venv/bin/dmypy"]
+    }
+
 Developing
 -------------
 

pylsp_mypy/plugin.py

@@ -165,6 +165,24 @@ def match_exclude_patterns(document_path: str, exclude_patterns: list) -> bool:
     return False
 
 
+def get_cmd(settings: Dict[str, Any], cmd: str) -> List[str]:
+    """Get the command to run from settings, falling back to searching the PATH.
+
+    If the command is not found in the settings and is not available on the PATH, an
+    empty list is returned.
+    """
+    command_key = f"{cmd}_command"
+    command: List[str] = settings.get(command_key, [])
+
+    if not command and shutil.which(cmd):
+        log.debug("'%s' not found in settings, using '%s' from PATH", command_key, cmd)
+        command = [cmd]
+
+    log.debug("Using %s command: %s", cmd, command)
+
+    return command
+
+
 @hookimpl
 def pylsp_lint(
     config: Config, workspace: Workspace, document: Document, is_saved: bool
@@ -304,12 +322,14 @@ def get_diagnostics(
         args.extend(["--incremental", "--follow-imports", settings.get("follow-imports", "silent")])
         args = apply_overrides(args, overrides)
 
-        if shutil.which("mypy"):
+        mypy_cmd = get_cmd(settings, "mypy")
+
+        if mypy_cmd:
             # mypy exists on path
             # -> use mypy on path
             log.info("executing mypy args = %s on path", args)
             completed_process = subprocess.run(
-                ["mypy", *args], capture_output=True, **windows_flag, encoding="utf-8"
+                [*mypy_cmd, *args], capture_output=True, **windows_flag, encoding="utf-8"
             )
             report = completed_process.stdout
             errors = completed_process.stderr
@@ -326,11 +346,13 @@ def get_diagnostics(
         # If daemon is dead/absent, kill will no-op.
         # In either case, reset to fresh state
 
-        if shutil.which("dmypy"):
+        dmypy_cmd = get_cmd(settings, "dmypy")
+
+        if dmypy_cmd:
             # dmypy exists on path
             # -> use dmypy on path
             completed_process = subprocess.run(
-                ["dmypy", "--status-file", dmypy_status_file, "status"],
+                [*dmypy_cmd, "--status-file", dmypy_status_file, "status"],
                 capture_output=True,
                 **windows_flag,
                 encoding="utf-8",
@@ -344,7 +366,7 @@ def get_diagnostics(
                     errors.strip(),
                 )
                 subprocess.run(
-                    ["dmypy", "--status-file", dmypy_status_file, "restart"],
+                    [*dmypy_cmd, "--status-file", dmypy_status_file, "restart"],
                     capture_output=True,
                     **windows_flag,
                     encoding="utf-8",
@@ -365,12 +387,12 @@ def get_diagnostics(
 
         # run to use existing daemon or restart if required
         args = ["--status-file", dmypy_status_file, "run", "--"] + apply_overrides(args, overrides)
-        if shutil.which("dmypy"):
+        if dmypy_cmd:
             # dmypy exists on path
             # -> use mypy on path
             log.info("dmypy run args = %s via path", args)
             completed_process = subprocess.run(
-                ["dmypy", *args], capture_output=True, **windows_flag, encoding="utf-8"
+                [*dmypy_cmd, *args], capture_output=True, **windows_flag, encoding="utf-8"
             )
             report = completed_process.stdout
             errors = completed_process.stderr

test/test_plugin.py

@@ -383,3 +383,89 @@ def test_config_exclude(tmpdir, workspace):
     workspace.update_config({"pylsp": {"plugins": {"pylsp_mypy": {"exclude": [exclude_path]}}}})
     diags = plugin.pylsp_lint(workspace._config, workspace, doc, is_saved=False)
     assert diags == []
+
+
+@pytest.mark.parametrize(
+    ("command", "settings", "cmd_on_path", "expected"),
+    [
+        ("mypy", {}, ["/bin/mypy"], ["mypy"]),
+        ("mypy", {}, None, []),
+        ("mypy", {"mypy_command": ["/path/to/mypy"]}, "/bin/mypy", ["/path/to/mypy"]),
+        ("mypy", {"mypy_command": ["/path/to/mypy"]}, None, ["/path/to/mypy"]),
+        ("dmypy", {}, "/bin/dmypy", ["dmypy"]),
+        ("dmypy", {}, None, []),
+        ("dmypy", {"dmypy_command": ["/path/to/dmypy"]}, "/bin/dmypy", ["/path/to/dmypy"]),
+        ("dmypy", {"dmypy_command": ["/path/to/dmypy"]}, None, ["/path/to/dmypy"]),
+    ],
+)
+def test_get_cmd(command, settings, cmd_on_path, expected):
+    with patch("shutil.which", return_value=cmd_on_path):
+        assert plugin.get_cmd(settings, command) == expected
+
+
+def test_config_overrides_mypy_command(last_diagnostics_monkeypatch, workspace):
+    last_diagnostics_monkeypatch.setattr(
+        FakeConfig,
+        "plugin_settings",
+        lambda _, p: (
+            {
+                "mypy_command": ["/path/to/mypy"],
+            }
+            if p == "pylsp_mypy"
+            else {}
+        ),
+    )
+
+    m = Mock(wraps=lambda a, **_: Mock(returncode=0, **{"stdout": ""}))
+    last_diagnostics_monkeypatch.setattr(plugin.subprocess, "run", m)
+
+    document = Document(DOC_URI, workspace, DOC_TYPE_ERR)
+
+    config = FakeConfig(uris.to_fs_path(workspace.root_uri))
+    plugin.pylsp_settings(config)
+
+    plugin.pylsp_lint(
+        config=config,
+        workspace=workspace,
+        document=document,
+        is_saved=False,
+    )
+
+    called_argv = m.call_args.args[0]
+    called_cmd = called_argv[0]
+    assert called_cmd == "/path/to/mypy"
+
+
+def test_config_overrides_dmypy_command(last_diagnostics_monkeypatch, workspace):
+    last_diagnostics_monkeypatch.setattr(
+        FakeConfig,
+        "plugin_settings",
+        lambda _, p: (
+            {
+                "dmypy": True,
+                "live_mode": False,
+                "dmypy_command": ["poetry", "run", "dmypy"],
+            }
+            if p == "pylsp_mypy"
+            else {}
+        ),
+    )
+
+    m = Mock(wraps=lambda a, **_: Mock(returncode=0, **{"stdout": ""}))
+    last_diagnostics_monkeypatch.setattr(plugin.subprocess, "run", m)
+
+    document = Document(DOC_URI, workspace, DOC_TYPE_ERR)
+
+    config = FakeConfig(uris.to_fs_path(workspace.root_uri))
+    plugin.pylsp_settings(config)
+
+    plugin.pylsp_lint(
+        config=config,
+        workspace=workspace,
+        document=document,
+        is_saved=False,
+    )
+
+    called_argv = m.call_args.args[0]
+    called_cmd = called_argv[:3]
+    assert called_cmd == ["poetry", "run", "dmypy"]