|
16 | 16 | from io import SEEK_CUR |
17 | 17 | from mmap import mmap |
18 | 18 | from os import DirEntry, path |
| 19 | +from pathlib import Path |
19 | 20 | from subprocess import PIPE, run |
20 | 21 | from types import ModuleType |
21 | | -from typing import Any, BinaryIO, Callable, Generator, Iterable, List, Optional |
| 22 | +from typing import ( |
| 23 | + Any, |
| 24 | + BinaryIO, |
| 25 | + Callable, |
| 26 | + Generator, |
| 27 | + Iterable, |
| 28 | + List, |
| 29 | + Optional, |
| 30 | + Union, |
| 31 | +) |
22 | 32 | from urllib.request import Request, urlopen |
23 | 33 |
|
24 | 34 | CHUNK_SIZE = 1024 * 1024 |
@@ -110,25 +120,78 @@ def executable_path(name: str) -> str: |
110 | 120 | return exe_path |
111 | 121 |
|
112 | 122 |
|
113 | | -def run_cmd(cmd: List[str], shell: bool = False): |
| 123 | +def _run_cmd( |
| 124 | + cmd: List[str], |
| 125 | + shell: bool = False, |
| 126 | + text: bool = True, |
| 127 | + data: Optional[Union[str, bytes]] = None, |
| 128 | + cwd: Optional[Path] = None, |
| 129 | +): |
| 130 | + if data is not None: |
| 131 | + if text: |
| 132 | + assert isinstance(data, str) |
| 133 | + else: |
| 134 | + assert isinstance(data, bytes) |
| 135 | + |
114 | 136 | cmd[0] = executable_path(cmd[0]) |
115 | 137 | proc = run( |
116 | 138 | cmd, |
117 | 139 | stdout=PIPE, |
118 | 140 | stderr=PIPE, |
119 | | - text=True, |
| 141 | + input=data, |
| 142 | + text=text, |
120 | 143 | shell=shell, |
121 | 144 | check=False, |
| 145 | + cwd=cwd, |
122 | 146 | ) |
123 | 147 | if proc.returncode != 0: |
124 | 148 | cmd_str = ' '.join(cmd) |
125 | 149 | s = f'Failed to run command "{cmd_str}":\n' |
126 | 150 | s += f'stdout:\n{proc.stdout}\n' |
127 | 151 | s += f'stderr:\n{proc.stderr}\n' |
128 | 152 | raise ValueError(s) |
| 153 | + |
| 154 | + if text: |
| 155 | + assert isinstance(proc.stdout, str) |
| 156 | + else: |
| 157 | + assert isinstance(proc.stdout, bytes) |
| 158 | + |
129 | 159 | return proc.stdout |
130 | 160 |
|
131 | 161 |
|
| 162 | +def run_cmd( |
| 163 | + cmd: List[str], |
| 164 | + shell: bool = False, |
| 165 | + data: Optional[str] = None, |
| 166 | + cwd: Optional[Path] = None, |
| 167 | +): |
| 168 | + output = _run_cmd( |
| 169 | + cmd, |
| 170 | + shell=shell, |
| 171 | + data=data, |
| 172 | + cwd=cwd, |
| 173 | + ) |
| 174 | + assert isinstance(output, str) |
| 175 | + return output |
| 176 | + |
| 177 | + |
| 178 | +def run_cmd_bytes( |
| 179 | + cmd: List[str], |
| 180 | + shell: bool = False, |
| 181 | + data: Optional[bytes] = None, |
| 182 | + cwd: Optional[Path] = None, |
| 183 | +): |
| 184 | + output = _run_cmd( |
| 185 | + cmd, |
| 186 | + shell=shell, |
| 187 | + data=data, |
| 188 | + cwd=cwd, |
| 189 | + text=False, |
| 190 | + ) |
| 191 | + assert isinstance(output, bytes) |
| 192 | + return output |
| 193 | + |
| 194 | + |
132 | 195 | def uncomment_line(line: str) -> Optional[str]: |
133 | 196 | line = line.strip() |
134 | 197 |
|
|
0 commit comments