|
6 | 6 | import subprocess |
7 | 7 | import sys |
8 | 8 | from pathlib import Path |
9 | | -from typing import IO, TextIO |
| 9 | +from typing import BinaryIO, TextIO, cast |
10 | 10 |
|
11 | 11 | import anyio |
12 | 12 | from anyio import to_thread |
@@ -58,18 +58,27 @@ class DummyProcess: |
58 | 58 |
|
59 | 59 | def __init__(self, popen_obj: subprocess.Popen[bytes]): |
60 | 60 | self.popen: subprocess.Popen[bytes] = popen_obj |
61 | | - self.stdin_raw: IO[bytes] | None = popen_obj.stdin |
62 | | - self.stdout_raw: IO[bytes] | None = popen_obj.stdout |
63 | | - self.stderr: IO[bytes] | None = popen_obj.stderr |
| 61 | + self.stdin_raw = popen_obj.stdin # type: ignore[assignment] |
| 62 | + self.stdout_raw = popen_obj.stdout # type: ignore[assignment] |
| 63 | + self.stderr = popen_obj.stderr # type: ignore[assignment] |
64 | 64 |
|
65 | | - self.stdin = FileWriteStream(self.stdin_raw) if self.stdin_raw else None |
66 | | - self.stdout = FileReadStream(self.stdout_raw) if self.stdout_raw else None |
| 65 | + self.stdin = ( |
| 66 | + FileWriteStream(cast(BinaryIO, self.stdin_raw)) if self.stdin_raw else None |
| 67 | + ) |
| 68 | + self.stdout = ( |
| 69 | + FileReadStream(cast(BinaryIO, self.stdout_raw)) if self.stdout_raw else None |
| 70 | + ) |
67 | 71 |
|
68 | 72 | async def __aenter__(self): |
69 | 73 | """Support async context manager entry.""" |
70 | 74 | return self |
71 | 75 |
|
72 | | - async def __aexit__(self, exc_type, exc_val, exc_tb): |
| 76 | + async def __aexit__( |
| 77 | + self, |
| 78 | + exc_type: BaseException | None, |
| 79 | + exc_val: BaseException | None, |
| 80 | + exc_tb: object | None, |
| 81 | + ) -> None: |
73 | 82 | """Terminate and wait on process exit inside a thread.""" |
74 | 83 | self.popen.terminate() |
75 | 84 | await to_thread.run_sync(self.popen.wait) |
@@ -122,11 +131,7 @@ async def create_windows_process( |
122 | 131 | env=env, |
123 | 132 | cwd=cwd, |
124 | 133 | bufsize=0, # Unbuffered output |
125 | | - creationflags=( |
126 | | - subprocess.CREATE_NO_WINDOW |
127 | | - if hasattr(subprocess, "CREATE_NO_WINDOW") |
128 | | - else 0 |
129 | | - ), |
| 134 | + creationflags = getattr(subprocess, "CREATE_NO_WINDOW", 0), |
130 | 135 | ) |
131 | 136 | return DummyProcess(popen_obj) |
132 | 137 |
|
|
0 commit comments