"""Server for mypy daemon mode.
This implements a daemon process which keeps useful state in memory
to enable fine-grained incremental reprocessing of changes.
"""
from __future__ import annotations
import argparse
import base64
import io
import json
import os
import pickle
import subprocess
import sys
import time
import traceback
from collections.abc import Sequence, Set as AbstractSet
from contextlib import redirect_stderr, redirect_stdout
from typing import Any, Callable, Final
from typing_extensions import TypeAlias as _TypeAlias
import mypy.build
import mypy.errors
import mypy.main
from mypy.dmypy_util import WriteToConn, receive, send
from mypy.find_sources import InvalidSourceList, create_source_list
from mypy.fscache import FileSystemCache
from mypy.fswatcher import FileData, FileSystemWatcher
from mypy.inspections import InspectionEngine
from mypy.ipc import IPCServer
from mypy.modulefinder import BuildSource, FindModuleCache, SearchPaths, compute_search_paths
from mypy.options import Options
from mypy.server.update import FineGrainedBuildManager, refresh_suppressed_submodules
from mypy.suggestions import SuggestionEngine, SuggestionFailure
from mypy.typestate import reset_global_state
from mypy.util import FancyFormatter, count_stats
from mypy.version import __version__
MEM_PROFILE: Final = False # If True, dump memory profile after initialization
if sys.platform == "win32":
from subprocess import STARTUPINFO
def daemonize(
options: Options, status_file: str, timeout: int | None = None, log_file: str | None = None
) -> int:
"""Create the daemon process via "dmypy daemon" and pass options via command line
When creating the daemon grandchild, we create it in a new console, which is
started hidden. We cannot use DETACHED_PROCESS since it will cause console windows
to pop up when starting. See
https://github.com/python/cpython/pull/4150#issuecomment-340215696
for more on why we can't have nice things.
It also pickles the options to be unpickled by mypy.
"""
command = [sys.executable, "-m", "mypy.dmypy", "--status-file", status_file, "daemon"]
pickled_options = pickle.dumps(options.snapshot())
command.append(f'--options-data="{base64.b64encode(pickled_options).decode()}"')
if timeout:
command.append(f"--timeout={timeout}")
if log_file:
command.append(f"--log-file={log_file}")
info = STARTUPINFO()
info.dwFlags = 0x1 # STARTF_USESHOWWINDOW aka use wShowWindow's value
info.wShowWindow = 0 # SW_HIDE aka make the window invisible
try:
subprocess.Popen(command, creationflags=0x10, startupinfo=info) # CREATE_NEW_CONSOLE
return 0
except subprocess.CalledProcessError as e:
return e.returncode
else:
def _daemonize_cb(func: Callable[[], None], log_file: str | None = None) -> int:
"""Arrange to call func() in a grandchild of the current process.
Return 0 for success, exit status for failure, negative if
subprocess killed by signal.
"""
# See https://stackoverflow.com/questions/473620/how-do-you-create-a-daemon-in-python
sys.stdout.flush()
sys.stderr.flush()
pid = os.fork()
if pid:
# Parent process: wait for child in case things go bad there.
npid, sts = os.waitpid(pid, 0)
sig = sts & 0xFF
if sig:
print("Child killed by signal", sig)
return -sig
sts = sts >> 8
if sts:
print("Child exit status", sts)
return sts
# Child process: do a bunch of UNIX stuff and then fork a grandchild.
try:
os.setsid() # Detach controlling terminal
os.umask(0o27)
devnull = os.open("/dev/null", os.O_RDWR)
os.dup2(devnull, 0)
os.dup2(devnull, 1)
os.dup2(devnull, 2)
os.close(devnull)
pid = os.fork()
if pid:
# Child is done, exit to parent.
os._exit(0)
# Grandchild: run the server.
if log_file:
sys.stdout = sys.stderr = open(log_file, "a", buffering=1)
fd = sys.stdout.fileno()
os.dup2(fd, 2)
os.dup2(fd, 1)
func()
finally:
# Make sure we never get back into the caller.
os._exit(1)
def daemonize(
options: Options, status_file: str, timeout: int | None = None, log_file: str | None = None
) -> int:
"""Run the mypy daemon in a grandchild of the current process
Return 0 for success, exit status for failure, negative if
subprocess killed by signal.
"""
return _daemonize_cb(Server(options, status_file, timeout).serve, log_file)
# Server code.
CONNECTION_NAME: Final = "dmypy"
def process_start_options(flags: list[str], allow_sources: bool) -> Options:
_, options = mypy.main.process_options(
["-i"] + flags, require_targets=False, server_options=True
)
if options.report_dirs:
print("dmypy: Ignoring report generation settings. Start/restart cannot generate reports.")
if options.junit_xml:
print(
"dmypy: Ignoring report generation settings. "
"Start/restart does not support --junit-xml. Pass it to check/recheck instead"
)
options.junit_xml = None
if not options.incremental:
sys.exit("dmypy: start/restart should not disable incremental mode")
if options.follow_imports not in ("skip", "error", "normal"):
sys.exit("dmypy: follow-imports=silent not supported")
return options
def ignore_suppressed_imports(module: str) -> bool:
"""Can we skip looking for newly unsuppressed imports to module?"""
# Various submodules of 'encodings' can be suppressed, since it
# uses module-level '__getattr__'. Skip them since there are many
# of them, and following imports to them is kind of pointless.
return module.startswith("encodings.")
ModulePathPair: _TypeAlias = tuple[str, str]
ModulePathPairs: _TypeAlias = list[ModulePathPair]
ChangesAndRemovals: _TypeAlias = tuple[ModulePathPairs, ModulePathPairs]
class Server:
# NOTE: the instance is constructed in the parent process but
# serve() is called in the grandchild (by daemonize()).
def __init__(self, options: Options, status_file: str, timeout: int | None = None) -> None:
"""Initialize the server with the desired mypy flags."""
self.options = options
# Snapshot the options info before we muck with it, to detect changes
self.options_snapshot = options.snapshot()
self.timeout = timeout
self.fine_grained_manager: FineGrainedBuildManager | None = None
if os.path.isfile(status_file):
os.unlink(status_file)
self.fscache = FileSystemCache()
options.raise_exceptions = True
options.incremental = True
options.fine_grained_incremental = True
options.show_traceback = True
if options.use_fine_grained_cache:
# Using fine_grained_cache implies generating and caring
# about the fine grained cache
options.cache_fine_grained = True
else:
options.cache_dir = os.devnull
# Fine-grained incremental doesn't support general partial types
# (details in https://github.com/python/mypy/issues/4492)
options.local_partial_types = True
self.status_file = status_file
# Since the object is created in the parent process we can check
# the output terminal options here.
self.formatter = FancyFormatter(sys.stdout, sys.stderr, options.hide_error_codes)
def _response_metadata(self) -> dict[str, str]:
py_version = f"{self.options.python_version[0]}_{self.options.python_version[1]}"
return {"platform": self.options.platform, "python_version": py_version}
def serve(self) -> None:
"""Serve requests, synchronously (no thread or fork)."""
command = None
server = IPCServer(CONNECTION_NAME, self.timeout)
orig_stdout = sys.stdout
orig_stderr = sys.stderr
try:
with open(self.status_file, "w") as f:
json.dump({"pid": os.getpid(), "connection_name": server.connection_name}, f)
f.write("\n") # I like my JSON with a trailing newline
while True:
with server:
data = receive(server)
sys.stdout = WriteToConn(server, "stdout", sys.stdout.isatty())
sys.stderr = WriteToConn(server, "stderr", sys.stderr.isatty())
resp: dict[str, Any] = {}
if "command" not in data:
resp = {"error": "No command found in request"}
else:
command = data["command"]
if not isinstance(command, str):
resp = {"error": "Command is not a string"}
else:
command = data.pop("command")
try:
resp = self.run_command(command, data)
except Exception:
# If we are crashing, report the crash to the client
tb = traceback.format_exception(*sys.exc_info())
resp = {"error": "Daemon crashed!\n" + "".join(tb)}
resp.update(self._response_metadata())
resp["final"] = True
send(server, resp)
raise
resp["final"] = True
try:
resp.update(self._response_metadata())
send(server, resp)
except OSError:
pass # Maybe the client hung up
if command == "stop":
reset_global_state()
sys.exit(0)
finally:
# Revert stdout/stderr so we can see any errors.
sys.stdout = orig_stdout
sys.stderr = orig_stderr
# If the final command is something other than a clean
# stop, remove the status file. (We can't just
# simplify the logic and always remove the file, since
# that could cause us to remove a future server's
# status file.)
if command != "stop":
os.unlink(self.status_file)
try:
server.cleanup() # try to remove the socket dir on Linux
except OSError:
pass
exc_info = sys.exc_info()
if exc_info[0] and exc_info[0] is not SystemExit:
traceback.print_exception(*exc_info)
def run_command(self, command: str, data: dict[str, object]) -> dict[str, object]:
"""Run a specific command from the registry."""
key = "cmd_" + command
method = getattr(self.__class__, key, None)
if method is None:
return {"error": f"Unrecognized command '{command}'"}
else:
if command not in {"check", "recheck", "run"}:
# Only the above commands use some error formatting.
del data["is_tty"]
del data["terminal_width"]
ret = method(self, **data)
assert isinstance(ret, dict)
return ret
# Command functions (run in the server via RPC).
def cmd_status(self, fswatcher_dump_file: str | None = None) -> dict[str, object]:
"""Return daemon status."""
res: dict[str, object] = {}
res.update(get_meminfo())
if fswatcher_dump_file:
data = self.fswatcher.dump_file_data() if hasattr(self, "fswatcher") else {}
# Using .dumps and then writing was noticeably faster than using dump
s = json.dumps(data)
with open(fswatcher_dump_file, "w") as f:
f.write(s)
return res
def cmd_stop(self) -> dict[str, object]:
"""Stop daemon."""
# We need to remove the status file *before* we complete the
# RPC. Otherwise a race condition exists where a subsequent
# command can see a status file from a dying server and think
# it is a live one.
os.unlink(self.status_file)
return {}
def cmd_run(
self,
version: str,
args: Sequence[str],
export_types: bool,
is_tty: bool,
terminal_width: int,
) -> dict[str, object]:
"""Check a list of files, triggering a restart if needed."""
stderr = io.StringIO()
stdout = io.StringIO()
try:
# Process options can exit on improper arguments, so we need to catch that and
# capture stderr so the client can report it
with redirect_stderr(stderr):
with redirect_stdout(stdout):
sources, options = mypy.main.process_options(
["-i"] + list(args),
require_targets=True,
server_options=True,
fscache=self.fscache,
program="mypy-daemon",
header=argparse.SUPPRESS,
)
# Signal that we need to restart if the options have changed
if not options.compare_stable(self.options_snapshot):
return {"restart": "configuration changed"}
if __version__ != version:
return {"restart": "mypy version changed"}
if self.fine_grained_manager:
manager = self.fine_grained_manager.manager
start_plugins_snapshot = manager.plugins_snapshot
_, current_plugins_snapshot = mypy.build.load_plugins(
options, manager.errors, sys.stdout, extra_plugins=()
)
if current_plugins_snapshot != start_plugins_snapshot:
return {"restart": "plugins changed"}
except InvalidSourceList as err:
return {"out": "", "err": str(err), "status": 2}
except SystemExit as e:
return {"out": stdout.getvalue(), "err": stderr.getvalue(), "status": e.code}
return self.check(sources, export_types, is_tty, terminal_width)
def cmd_check(
self, files: Sequence[str], export_types: bool, is_tty: bool, terminal_width: int
) -> dict[str, object]:
"""Check a list of files."""
try:
sources = create_source_list(files, self.options, self.fscache)
except InvalidSourceList as err:
return {"out": "", "err": str(err), "status": 2}
return self.check(sources, export_types, is_tty, terminal_width)
def cmd_recheck(
self,
is_tty: bool,
terminal_width: int,
export_types: bool,
remove: list[str] | None = None,
update: list[str] | None = None,
) -> dict[str, object]:
"""Check the same list of files we checked most recently.
If remove/update is given, they modify the previous list;
if all are None, stat() is called for each file in the previous list.
"""
t0 = time.time()
if not self.fine_grained_manager:
return {"error": "Command 'recheck' is only valid after a 'check' command"}
sources = self.previous_sources
if remove:
removals = set(remove)
sources = [s for s in sources if s.path and s.path not in removals]
if update:
# Sort list of file updates by extension, so *.pyi files are first.
update.sort(key=lambda f: os.path.splitext(f)[1], reverse=True)
known = {s.path for s in sources if s.path}
added = [p for p in update if p not in known]
try:
added_sources = create_source_list(added, self.options, self.fscache)
except InvalidSourceList as err:
return {"out": "", "err": str(err), "status": 2}
sources = sources + added_sources # Make a copy!
t1 = time.time()
manager = self.fine_grained_manager.manager
manager.log(f"fine-grained increment: cmd_recheck: {t1 - t0:.3f}s")
old_export_types = self.options.export_types
self.options.export_types = self.options.export_types or export_types
if not self.following_imports():
messages = self.fine_grained_increment(
sources, remove, update, explicit_export_types=export_types
)
else:
assert remove is None and update is None
messages = self.fine_grained_increment_follow_imports(
sources, explicit_export_types=export_types
)
res = self.increment_output(messages, sources, is_tty, terminal_width)
self.flush_caches()
self.update_stats(res)
self.options.export_types = old_export_types
return res
def check(
self, sources: list[BuildSource], export_types: bool, is_tty: bool, terminal_width: int
) -> dict[str, Any]:
"""Check using fine-grained incremental mode.
If is_tty is True format the output nicely with colors and summary line
(unless disabled in self.options). Also pass the terminal_width to formatter.
"""
old_export_types = self.options.export_types
self.options.export_types = self.options.export_types or export_types
if not self.fine_grained_manager:
res = self.initialize_fine_grained(sources, is_tty, terminal_width)
else:
if not self.following_imports():
messages = self.fine_grained_increment(sources, explicit_export_types=export_types)
else:
messages = self.fine_grained_increment_follow_imports(
sources, explicit_export_types=export_types
)
res = self.increment_output(messages, sources, is_tty, terminal_width)
self.flush_caches()
self.update_stats(res)
self.options.export_types = old_export_types
return res
def flush_caches(self) -> None:
self.fscache.flush()
if self.fine_grained_manager:
self.fine_grained_manager.flush_cache()
def update_stats(self, res: dict[str, Any]) -> None:
if self.fine_grained_manager:
manager = self.fine_grained_manager.manager
manager.dump_stats()
res["stats"] = manager.stats
manager.stats = {}
def following_imports(self) -> bool:
"""Are we following imports?"""
# TODO: What about silent?
return self.options.follow_imports == "normal"
def initialize_fine_grained(
self, sources: list[BuildSource], is_tty: bool, terminal_width: int
) -> dict[str, Any]:
self.fswatcher = FileSystemWatcher(self.fscache)
t0 = time.time()
self.update_sources(sources)
t1 = time.time()
try:
result = mypy.build.build(sources=sources, options=self.options, fscache=self.fscache)
except mypy.errors.CompileError as e:
output = "".join(s + "\n" for s in e.messages)
if e.use_stdout:
out, err = output, ""
else:
out, err = "", output
return {"out": out, "err": err, "status": 2}
messages = result.errors
self.fine_grained_manager = FineGrainedBuildManager(result)
original_sources_len = len(sources)
if self.following_imports():
sources = find_all_sources_in_build(self.fine_grained_manager.graph, sources)
self.update_sources(sources)
self.previous_sources = sources
# If we are using the fine-grained cache, build hasn't actually done
# the typechecking on the updated files yet.
# Run a fine-grained update starting from the cached data
if result.used_cache:
t2 = time.time()
# Pull times and hashes out of the saved_cache and stick them into
# the fswatcher, so we pick up the changes.
for state in self.fine_grained_manager.graph.values():
meta = state.meta
if meta is None:
continue
assert state.path is not None
self.fswatcher.set_file_data(
state.path,
FileData(st_mtime=float(meta.mtime), st_size=meta.size, hash=meta.hash),
)
changed, removed = self.find_changed(sources)
changed += self.find_added_suppressed(
self.fine_grained_manager.graph,
set(),
self.fine_grained_manager.manager.search_paths,
)
# Find anything that has had its dependency list change
for state in self.fine_grained_manager.graph.values():
if not state.is_fresh():
assert state.path is not None
changed.append((state.id, state.path))
t3 = time.time()
# Run an update
messages = self.fine_grained_manager.update(changed, removed)
if self.following_imports():
# We need to do another update to any new files found by following imports.
messages = self.fine_grained_increment_follow_imports(sources)
t4 = time.time()
self.fine_grained_manager.manager.add_stats(
update_sources_time=t1 - t0,
build_time=t2 - t1,
find_changes_time=t3 - t2,
fg_update_time=t4 - t3,
files_changed=len(removed) + len(changed),
)
else:
# Stores the initial state of sources as a side effect.
self.fswatcher.find_changed()
if MEM_PROFILE:
from mypy.memprofile import print_memory_profile
print_memory_profile(run_gc=False)
__, n_notes, __ = count_stats(messages)
status = 1 if messages and n_notes < len(messages) else 0
# We use explicit sources length to match the logic in non-incremental mode.
messages = self.pretty_messages(messages, original_sources_len, is_tty, terminal_width)
return {"out": "".join(s + "\n" for s in messages), "err": "", "status": status}
def fine_grained_increment(
self,
sources: list[BuildSource],
remove: list[str] | None = None,
update: list[str] | None = None,
explicit_export_types: bool = False,
) -> list[str]:
"""Perform a fine-grained type checking increment.
If remove and update are None, determine changed paths by using
fswatcher. Otherwise, assume that only these files have changes.
Args:
sources: sources passed on the command line
remove: paths of files that have been removed
update: paths of files that have been changed or created
explicit_export_types: --export-type was passed in a check command
(as opposite to being set in dmypy start)
"""
assert self.fine_grained_manager is not None
manager = self.fine_grained_manager.manager
t0 = time.time()
if remove is None and update is None:
# Use the fswatcher to determine which files were changed
# (updated or added) or removed.
self.update_sources(sources)
changed, removed = self.find_changed(sources)
else:
# Use the remove/update lists to update fswatcher.
# This avoids calling stat() for unchanged files.
changed, removed = self.update_changed(sources, remove or [], update or [])
if explicit_export_types:
# If --export-types is given, we need to force full re-checking of all
# explicitly passed files, since we need to visit each expression.
add_all_sources_to_changed(sources, changed)
changed += self.find_added_suppressed(
self.fine_grained_manager.graph, set(), manager.search_paths
)
manager.search_paths = compute_search_paths(sources, manager.options, manager.data_dir)
t1 = time.time()
manager.log(f"fine-grained increment: find_changed: {t1 - t0:.3f}s")
messages = self.fine_grained_manager.update(changed, removed)
t2 = time.time()
manager.log(f"fine-grained increment: update: {t2 - t1:.3f}s")
manager.add_stats(
find_changes_time=t1 - t0,
fg_update_time=t2 - t1,
files_changed=len(removed) + len(changed),
)
self.previous_sources = sources
return messages
def fine_grained_increment_follow_imports(
self, sources: list[BuildSource], explicit_export_types: bool = False
) -> list[str]:
"""Like fine_grained_increment, but follow imports."""
t0 = time.time()
# TODO: Support file events
assert self.fine_grained_manager is not None
fine_grained_manager = self.fine_grained_manager
graph = fine_grained_manager.graph
manager = fine_grained_manager.manager
orig_modules = list(graph.keys())
self.update_sources(sources)
changed_paths = self.fswatcher.find_changed()
manager.search_paths = compute_search_paths(sources, manager.options, manager.data_dir)
t1 = time.time()
manager.log(f"fine-grained increment: find_changed: {t1 - t0:.3f}s")
seen = {source.module for source in sources}
# Find changed modules reachable from roots (or in roots) already in graph.
changed, new_files = self.find_reachable_changed_modules(
sources, graph, seen, changed_paths
)
# Same as in fine_grained_increment().
self.add_explicitly_new(sources, changed)
if explicit_export_types:
# Same as in fine_grained_increment().
add_all_sources_to_changed(sources, changed)
sources.extend(new_files)
# Process changes directly reachable from roots.
messages = fine_grained_manager.update(changed, [], followed=True)
# Follow deps from changed modules (still within graph).
worklist = changed.copy()
while worklist:
module = worklist.pop()
if module[0] not in graph:
continue
sources2 = self.direct_imports(module, graph)
# Filter anything already seen before. This prevents
# infinite looping if there are any self edges. (Self
# edges are maybe a bug, but...)
sources2 = [source for source in sources2 if source.module not in seen]
changed, new_files = self.find_reachable_changed_modules(
sources2, graph, seen, changed_paths
)
self.update_sources(new_files)
messages = fine_grained_manager.update(changed, [], followed=True)
worklist.extend(changed)
t2 = time.time()
def refresh_file(module: str, path: str) -> list[str]:
return fine_grained_manager.update([(module, path)], [], followed=True)
for module_id, state in list(graph.items()):
new_messages = refresh_suppressed_submodules(
module_id, state.path, fine_grained_manager.deps, graph, self.fscache, refresh_file
)
if new_messages is not None:
messages = new_messages
t3 = time.time()
# There may be new files that became available, currently treated as
# suppressed imports. Process them.
while True:
new_unsuppressed = self.find_added_suppressed(graph, seen, manager.search_paths)
if not new_unsuppressed:
break
new_files = [BuildSource(mod[1], mod[0], followed=True) for mod in new_unsuppressed]
sources.extend(new_files)
self.update_sources(new_files)
messages = fine_grained_manager.update(new_unsuppressed, [], followed=True)
for module_id, path in new_unsuppressed:
new_messages = refresh_suppressed_submodules(
module_id, path, fine_grained_manager.deps, graph, self.fscache, refresh_file
)
if new_messages is not None:
messages = new_messages
t4 = time.time()
# Find all original modules in graph that were not reached -- they are deleted.
to_delete = []
for module_id in orig_modules:
if module_id not in graph:
continue
if module_id not in seen:
module_path = graph[module_id].path
assert module_path is not None
to_delete.append((module_id, module_path))
if to_delete:
messages = fine_grained_manager.update([], to_delete)
fix_module_deps(graph)
self.previous_sources = find_all_sources_in_build(graph)
self.update_sources(self.previous_sources)
# Store current file state as side effect
self.fswatcher.find_changed()
t5 = time.time()
manager.log(f"fine-grained increment: update: {t5 - t1:.3f}s")
manager.add_stats(
find_changes_time=t1 - t0,
fg_update_time=t2 - t1,
refresh_suppressed_time=t3 - t2,
find_added_suppressed_time=t4 - t3,
cleanup_time=t5 - t4,
)
return messages
def find_reachable_changed_modules(
self,
roots: list[BuildSource],
graph: mypy.build.Graph,
seen: set[str],
changed_paths: AbstractSet[str],
) -> tuple[list[tuple[str, str]], list[BuildSource]]:
"""Follow imports within graph from given sources until hitting changed modules.
If we find a changed module, we can't continue following imports as the imports
may have changed.
Args:
roots: modules where to start search from
graph: module graph to use for the search
seen: modules we've seen before that won't be visited (mutated here!!)
changed_paths: which paths have changed (stop search here and return any found)
Return (encountered reachable changed modules,
unchanged files not in sources_set traversed).
"""
changed = []
new_files = []
worklist = roots.copy()
seen.update(source.module for source in worklist)
while worklist:
nxt = worklist.pop()
if nxt.module not in seen:
seen.add(nxt.module)
new_files.append(nxt)
if nxt.path in changed_paths:
assert nxt.path is not None # TODO
changed.append((nxt.module, nxt.path))
elif nxt.module in graph:
state = graph[nxt.module]
for dep in state.dependencies:
if dep not in seen:
seen.add(dep)
worklist.append(BuildSource(graph[dep].path, graph[dep].id, followed=True))
return changed, new_files
def direct_imports(
self, module: tuple[str, str], graph: mypy.build.Graph
) -> list[BuildSource]:
"""Return the direct imports of module not included in seen."""
state = graph[module[0]]
return [BuildSource(graph[dep].path, dep, followed=True) for dep in state.dependencies]
def find_added_suppressed(
self, graph: mypy.build.Graph, seen: set[str], search_paths: SearchPaths
) -> list[tuple[str, str]]:
"""Find suppressed modules that have been added (and not included in seen).
Args:
seen: reachable modules we've seen before (mutated here!!)
Return suppressed, added modules.
"""
all_suppressed = set()
for state in graph.values():
all_suppressed |= state.suppressed_set
# Filter out things that shouldn't actually be considered suppressed.
#
# TODO: Figure out why these are treated as suppressed
all_suppressed = {
module
for module in all_suppressed
if module not in graph and not ignore_suppressed_imports(module)
}
# Optimization: skip top-level packages that are obviously not
# there, to avoid calling the relatively slow find_module()
# below too many times.
packages = {module.split(".", 1)[0] for module in all_suppressed}
packages = filter_out_missing_top_level_packages(packages, search_paths, self.fscache)
# TODO: Namespace packages
finder = FindModuleCache(search_paths, self.fscache, self.options)
found = []
for module in all_suppressed:
top_level_pkg = module.split(".", 1)[0]
if top_level_pkg not in packages:
# Fast path: non-existent top-level package
continue
result = finder.find_module(module, fast_path=True)
if isinstance(result, str) and module not in seen:
# When not following imports, we only follow imports to .pyi files.
if not self.following_imports() and not result.endswith(".pyi"):
continue
found.append((module, result))
seen.add(module)
return found
def increment_output(
self, messages: list[str], sources: list[BuildSource], is_tty: bool, terminal_width: int
) -> dict[str, Any]:
status = 1 if messages else 0
messages = self.pretty_messages(messages, len(sources), is_tty, terminal_width)
return {"out": "".join(s + "\n" for s in messages), "err": "", "status": status}
def pretty_messages(
self,
messages: list[str],
n_sources: int,
is_tty: bool = False,
terminal_width: int | None = None,
) -> list[str]:
use_color = self.options.color_output and is_tty
fit_width = self.options.pretty and is_tty
if fit_width:
messages = self.formatter.fit_in_terminal(
messages, fixed_terminal_width=terminal_width
)
if self.options.error_summary:
summary: str | None = None
n_errors, n_notes, n_files = count_stats(messages)
if n_errors:
summary = self.formatter.format_error(
n_errors, n_files, n_sources, use_color=use_color
)
elif not messages or n_notes == len(messages):
summary = self.formatter.format_success(n_sources, use_color)
if summary:
# Create new list to avoid appending multiple summaries on successive runs.
messages = messages + [summary]
if use_color:
messages = [self.formatter.colorize(m) for m in messages]
return messages
def update_sources(self, sources: list[BuildSource]) -> None:
paths = [source.path for source in sources if source.path is not None]
if self.following_imports():
# Filter out directories (used for namespace packages).
paths = [path for path in paths if self.fscache.isfile(path)]
self.fswatcher.add_watched_paths(paths)
def update_changed(
self, sources: list[BuildSource], remove: list[str], update: list[str]
) -> ChangesAndRemovals:
changed_paths = self.fswatcher.update_changed(remove, update)
return self._find_changed(sources, changed_paths)
def find_changed(self, sources: list[BuildSource]) -> ChangesAndRemovals:
changed_paths = self.fswatcher.find_changed()
return self._find_changed(sources, changed_paths)
def _find_changed(
self, sources: list[BuildSource], changed_paths: AbstractSet[str]
) -> ChangesAndRemovals:
# Find anything that has been added or modified
changed = [
(source.module, source.path)
for source in sources
if source.path and source.path in changed_paths
]
# Now find anything that has been removed from the build
modules = {source.module for source in sources}
omitted = [source for source in self.previous_sources if source.module not in modules]
removed = []
for source in omitted:
path = source.path
assert path
removed.append((source.module, path))
self.add_explicitly_new(sources, changed)
# Find anything that has had its module path change because of added or removed __init__s
last = {s.path: s.module for s in self.previous_sources}
for s in sources:
assert s.path
if s.path in last and last[s.path] != s.module:
# Mark it as removed from its old name and changed at its new name
removed.append((last[s.path], s.path))
changed.append((s.module, s.path))
return changed, removed
def add_explicitly_new(
self, sources: list[BuildSource], changed: list[tuple[str, str]]
) -> None:
# Always add modules that were (re-)added, since they may be detected as not changed by
# fswatcher (if they were actually not changed), but they may still need to be checked
# in case they had errors before they were deleted from sources on previous runs.
previous_modules = {source.module for source in self.previous_sources}
changed_set = set(changed)
changed.extend(
[
(source.module, source.path)
for source in sources
if source.path
and source.module not in previous_modules
and (source.module, source.path) not in changed_set
]
)
def cmd_inspect(
self,
show: str,
location: str,
verbosity: int = 0,
limit: int = 0,
include_span: bool = False,
include_kind: bool = False,
include_object_attrs: bool = False,
union_attrs: bool = False,
force_reload: bool = False,
) -> dict[str, object]:
"""Locate and inspect expression(s)."""
if not self.fine_grained_manager:
return {
"error": 'Command "inspect" is only valid after a "check" command'
" (that produces no parse errors)"
}
engine = InspectionEngine(
self.fine_grained_manager,
verbosity=verbosity,
limit=limit,
include_span=include_span,
include_kind=include_kind,
include_object_attrs=include_object_attrs,
union_attrs=union_attrs,
force_reload=force_reload,
)
old_inspections = self.options.inspections
self.options.inspections = True
try:
if show == "type":
result = engine.get_type(location)
elif show == "attrs":
result = engine.get_attrs(location)
elif show == "definition":
result = engine.get_definition(location)
else:
assert False, "Unknown inspection kind"
finally:
self.options.inspections = old_inspections
if "out" in result:
assert isinstance(result["out"], str)
result["out"] += "\n"
return result
def cmd_suggest(self, function: str, callsites: bool, **kwargs: Any) -> dict[str, object]:
"""Suggest a signature for a function."""
if not self.fine_grained_manager:
return {
"error": "Command 'suggest' is only valid after a 'check' command"
" (that produces no parse errors)"
}
engine = SuggestionEngine(self.fine_grained_manager, **kwargs)
try:
if callsites:
out = engine.suggest_callsites(function)
else:
out = engine.suggest(function)
except SuggestionFailure as err:
return {"error": str(err)}
else:
if not out:
out = "No suggestions\n"
elif not out.endswith("\n"):
out += "\n"
return {"out": out, "err": "", "status": 0}
finally:
self.flush_caches()
def cmd_hang(self) -> dict[str, object]:
"""Hang for 100 seconds, as a debug hack."""
time.sleep(100)
return {}
# Misc utilities.
MiB: Final = 2**20
def get_meminfo() -> dict[str, Any]:
res: dict[str, Any] = {}
try:
import psutil
except ImportError:
res["memory_psutil_missing"] = (
"psutil not found, run pip install mypy[dmypy] "
"to install the needed components for dmypy"
)
else:
process = psutil.Process()
meminfo = process.memory_info()
res["memory_rss_mib"] = meminfo.rss / MiB
res["memory_vms_mib"] = meminfo.vms / MiB
if sys.platform == "win32":
res["memory_maxrss_mib"] = meminfo.peak_wset / MiB
else:
# See https://stackoverflow.com/questions/938733/total-memory-used-by-python-process
import resource # Since it doesn't exist on Windows.
rusage = resource.getrusage(resource.RUSAGE_SELF)
if sys.platform == "darwin":
factor = 1
else:
factor = 1024 # Linux
res["memory_maxrss_mib"] = rusage.ru_maxrss * factor / MiB
return res
def find_all_sources_in_build(
graph: mypy.build.Graph, extra: Sequence[BuildSource] = ()
) -> list[BuildSource]:
result = list(extra)
seen = {source.module for source in result}
for module, state in graph.items():
if module not in seen:
result.append(BuildSource(state.path, module))
return result
def add_all_sources_to_changed(sources: list[BuildSource], changed: list[tuple[str, str]]) -> None:
"""Add all (explicit) sources to the list changed files in place.
Use this when re-processing of unchanged files is needed (e.g. for
the purpose of exporting types for inspections).
"""
changed_set = set(changed)
changed.extend(
[
(bs.module, bs.path)
for bs in sources
if bs.path and (bs.module, bs.path) not in changed_set
]
)
def fix_module_deps(graph: mypy.build.Graph) -> None:
"""After an incremental update, update module dependencies to reflect the new state.
This can make some suppressed dependencies non-suppressed, and vice versa (if modules
have been added to or removed from the build).
"""
for state in graph.values():
new_suppressed = []
new_dependencies = []
for dep in state.dependencies + state.suppressed:
if dep in graph:
new_dependencies.append(dep)
else:
new_suppressed.append(dep)
state.dependencies = new_dependencies
state.dependencies_set = set(new_dependencies)
state.suppressed = new_suppressed
state.suppressed_set = set(new_suppressed)
def filter_out_missing_top_level_packages(
packages: set[str], search_paths: SearchPaths, fscache: FileSystemCache
) -> set[str]:
"""Quickly filter out obviously missing top-level packages.
Return packages with entries that can't be found removed.
This is approximate: some packages that aren't actually valid may be
included. However, all potentially valid packages must be returned.
"""
# Start with a empty set and add all potential top-level packages.
found = set()
paths = (
search_paths.python_path
+ search_paths.mypy_path
+ search_paths.package_path
+ search_paths.typeshed_path
)
for p in paths:
try:
entries = fscache.listdir(p)
except Exception:
entries = []
for entry in entries:
# The code is hand-optimized for mypyc since this may be somewhat
# performance-critical.
if entry.endswith(".py"):
entry = entry[:-3]
elif entry.endswith(".pyi"):
entry = entry[:-4]
elif entry.endswith("-stubs"):
# Possible PEP 561 stub package
entry = entry[:-6]
if entry in packages:
found.add(entry)
return found
