118 lines
2.9 KiB
Python
118 lines
2.9 KiB
Python
"""Shared code between dmypy.py and dmypy_server.py.
|
|
|
|
This should be pretty lightweight and not depend on other mypy code (other than ipc).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
from collections.abc import Iterable, Iterator
|
|
from types import TracebackType
|
|
from typing import Any, Final, TextIO
|
|
|
|
from mypy.ipc import IPCBase
|
|
|
|
DEFAULT_STATUS_FILE: Final = ".dmypy.json"
|
|
|
|
|
|
def receive(connection: IPCBase) -> Any:
|
|
"""Receive single JSON data frame from a connection.
|
|
|
|
Raise OSError if the data received is not valid JSON or if it is
|
|
not a dict.
|
|
"""
|
|
bdata = connection.read()
|
|
if not bdata:
|
|
raise OSError("No data received")
|
|
try:
|
|
data = json.loads(bdata)
|
|
except Exception as e:
|
|
raise OSError("Data received is not valid JSON") from e
|
|
if not isinstance(data, dict):
|
|
raise OSError(f"Data received is not a dict ({type(data)})")
|
|
return data
|
|
|
|
|
|
def send(connection: IPCBase, data: Any) -> None:
|
|
"""Send data to a connection encoded and framed.
|
|
|
|
The data must be JSON-serializable. We assume that a single send call is a
|
|
single frame to be sent on the connect.
|
|
"""
|
|
connection.write(json.dumps(data))
|
|
|
|
|
|
class WriteToConn(TextIO):
|
|
"""Helper class to write to a connection instead of standard output."""
|
|
|
|
def __init__(self, server: IPCBase, output_key: str, isatty: bool) -> None:
|
|
self.server = server
|
|
self.output_key = output_key
|
|
self._isatty = isatty
|
|
|
|
def __enter__(self) -> TextIO:
|
|
return self
|
|
|
|
def __exit__(
|
|
self,
|
|
t: type[BaseException] | None,
|
|
value: BaseException | None,
|
|
traceback: TracebackType | None,
|
|
) -> None:
|
|
pass
|
|
|
|
def __iter__(self) -> Iterator[str]:
|
|
raise io.UnsupportedOperation
|
|
|
|
def __next__(self) -> str:
|
|
raise io.UnsupportedOperation
|
|
|
|
def close(self) -> None:
|
|
pass
|
|
|
|
def fileno(self) -> int:
|
|
raise OSError
|
|
|
|
def flush(self) -> None:
|
|
pass
|
|
|
|
def isatty(self) -> bool:
|
|
return self._isatty
|
|
|
|
def read(self, n: int = 0) -> str:
|
|
raise io.UnsupportedOperation
|
|
|
|
def readable(self) -> bool:
|
|
return False
|
|
|
|
def readline(self, limit: int = 0) -> str:
|
|
raise io.UnsupportedOperation
|
|
|
|
def readlines(self, hint: int = 0) -> list[str]:
|
|
raise io.UnsupportedOperation
|
|
|
|
def seek(self, offset: int, whence: int = 0) -> int:
|
|
raise io.UnsupportedOperation
|
|
|
|
def seekable(self) -> bool:
|
|
return False
|
|
|
|
def tell(self) -> int:
|
|
raise io.UnsupportedOperation
|
|
|
|
def truncate(self, size: int | None = 0) -> int:
|
|
raise io.UnsupportedOperation
|
|
|
|
def write(self, output: str) -> int:
|
|
resp: dict[str, Any] = {self.output_key: output}
|
|
send(self.server, resp)
|
|
return len(output)
|
|
|
|
def writable(self) -> bool:
|
|
return True
|
|
|
|
def writelines(self, lines: Iterable[str]) -> None:
|
|
for s in lines:
|
|
self.write(s)
|