| 170 | |
| 171 | |
| 172 | class Server: |
| 173 | # NOTE: the instance is constructed in the parent process but |
| 174 | # serve() is called in the grandchild (by daemonize()). |
| 175 | |
| 176 | def __init__(self, options: Options, status_file: str, timeout: int | None = None) -> None: |
| 177 | """Initialize the server with the desired mypy flags.""" |
| 178 | self.options = options |
| 179 | # Snapshot the options info before we muck with it, to detect changes |
| 180 | self.options_snapshot = options.snapshot() |
| 181 | self.timeout = timeout |
| 182 | self.fine_grained_manager: FineGrainedBuildManager | None = None |
| 183 | |
| 184 | if os.path.isfile(status_file): |
| 185 | os.unlink(status_file) |
| 186 | |
| 187 | self.fscache = FileSystemCache() |
| 188 | |
| 189 | options.raise_exceptions = True |
| 190 | options.incremental = True |
| 191 | options.fine_grained_incremental = True |
| 192 | options.show_traceback = True |
| 193 | if options.use_fine_grained_cache: |
| 194 | # Using fine_grained_cache implies generating and caring |
| 195 | # about the fine grained cache |
| 196 | options.cache_fine_grained = True |
| 197 | else: |
| 198 | options.cache_dir = os.devnull |
| 199 | # Fine-grained incremental doesn't support general partial types |
| 200 | # (details in https://github.com/python/mypy/issues/4492) |
| 201 | options.local_partial_types = True |
| 202 | self.status_file = status_file |
| 203 | |
| 204 | # Since the object is created in the parent process we can check |
| 205 | # the output terminal options here. |
| 206 | self.formatter = FancyFormatter(sys.stdout, sys.stderr, options.hide_error_codes) |
| 207 | |
| 208 | def _response_metadata(self) -> dict[str, str]: |
| 209 | py_version = f"{self.options.python_version[0]}_{self.options.python_version[1]}" |
| 210 | return {"platform": self.options.platform, "python_version": py_version} |
| 211 | |
| 212 | def serve(self) -> None: |
| 213 | """Serve requests, synchronously (no thread or fork).""" |
| 214 | |
| 215 | command = None |
| 216 | server = IPCServer(CONNECTION_NAME, self.timeout) |
| 217 | orig_stdout = sys.stdout |
| 218 | orig_stderr = sys.stderr |
| 219 | |
| 220 | try: |
| 221 | with open(self.status_file, "w") as f: |
| 222 | json.dump({"pid": os.getpid(), "connection_name": server.connection_name}, f) |
| 223 | f.write("\n") # I like my JSON with a trailing newline |
| 224 | while True: |
| 225 | with server: |
| 226 | data = receive(server) |
| 227 | sys.stdout = WriteToConn(server, "stdout", sys.stdout.isatty()) |
| 228 | sys.stderr = WriteToConn(server, "stderr", sys.stderr.isatty()) |
| 229 | resp: dict[str, Any] = {} |