dmypy_util.py 846 B

12345678910111213141516171819202122232425262728293031
  1. """Shared code between dmypy.py and dmypy_server.py.
  2. This should be pretty lightweight and not depend on other mypy code (other than ipc).
  3. """
  4. from __future__ import annotations
  5. import json
  6. from typing import Any, Final
  7. from mypy.ipc import IPCBase
  8. DEFAULT_STATUS_FILE: Final = ".dmypy.json"
  9. def receive(connection: IPCBase) -> Any:
  10. """Receive JSON data from a connection until EOF.
  11. Raise OSError if the data received is not valid JSON or if it is
  12. not a dict.
  13. """
  14. bdata = connection.read()
  15. if not bdata:
  16. raise OSError("No data received")
  17. try:
  18. data = json.loads(bdata.decode("utf8"))
  19. except Exception as e:
  20. raise OSError("Data received is not valid JSON") from e
  21. if not isinstance(data, dict):
  22. raise OSError(f"Data received is not a dict ({type(data)})")
  23. return data