| 21 | |
| 22 | @dataclasses.dataclass(slots=True, frozen=True) |
| 23 | class JsonFile: |
| 24 | # file type depends on file_type: |
| 25 | # - UNIX_FD: file descriptor (int) |
| 26 | # - WINDOWS_HANDLE: handle (int) |
| 27 | # - STDOUT: use process stdout (None) |
| 28 | file: int | None |
| 29 | file_type: str |
| 30 | |
| 31 | def configure_subprocess(self, popen_kwargs: dict[str, Any]) -> None: |
| 32 | match self.file_type: |
| 33 | case JsonFileType.UNIX_FD: |
| 34 | # Unix file descriptor |
| 35 | popen_kwargs['pass_fds'] = [self.file] |
| 36 | case JsonFileType.WINDOWS_HANDLE: |
| 37 | # Windows handle |
| 38 | # We run mypy with `--platform=linux` so it complains about this: |
| 39 | startupinfo = subprocess.STARTUPINFO() # type: ignore[attr-defined] |
| 40 | startupinfo.lpAttributeList = {"handle_list": [self.file]} |
| 41 | popen_kwargs['startupinfo'] = startupinfo |
| 42 | |
| 43 | @contextlib.contextmanager |
| 44 | def inherit_subprocess(self) -> Iterator[None]: |
| 45 | if sys.platform == 'win32' and self.file_type == JsonFileType.WINDOWS_HANDLE: |
| 46 | os.set_handle_inheritable(self.file, True) |
| 47 | try: |
| 48 | yield |
| 49 | finally: |
| 50 | os.set_handle_inheritable(self.file, False) |
| 51 | else: |
| 52 | yield |
| 53 | |
| 54 | def open(self, mode='r', *, encoding): |
| 55 | if self.file_type == JsonFileType.STDOUT: |
| 56 | raise ValueError("for STDOUT file type, just use sys.stdout") |
| 57 | |
| 58 | file = self.file |
| 59 | if self.file_type == JsonFileType.WINDOWS_HANDLE: |
| 60 | import msvcrt |
| 61 | # Create a file descriptor from the handle |
| 62 | file = msvcrt.open_osfhandle(file, os.O_WRONLY) |
| 63 | return open(file, mode, encoding=encoding) |
| 64 | |
| 65 | |
| 66 | @dataclasses.dataclass(slots=True, frozen=True) |
no outgoing calls
searching dependent graphs…