|
5 | 5 | import os |
6 | 6 | import subprocess |
7 | 7 | import sys |
8 | | -from typing import Callable, Optional, Union |
| 8 | +from typing import Optional, Protocol, Union |
9 | 9 |
|
10 | 10 | from .. import exc |
11 | 11 | from ..types import StrOrBytesPath |
@@ -127,13 +127,21 @@ def process(self, msg, kwargs): |
127 | 127 | return msg, kwargs |
128 | 128 |
|
129 | 129 |
|
| 130 | +class ProgressCallbackProtocol(Protocol): |
| 131 | + """Callback to report subprocess communication.""" |
| 132 | + |
| 133 | + def __call__(self, output: Union[str, bytes], timestamp: datetime.datetime): |
| 134 | + """Callback signature for subprocess communication.""" |
| 135 | + ... |
| 136 | + |
| 137 | + |
130 | 138 | def run( |
131 | 139 | cmd: Union[str, list[str]], |
132 | 140 | shell: bool = False, |
133 | 141 | cwd: Optional[StrOrBytesPath] = None, |
134 | 142 | log_in_real_time: bool = True, |
135 | 143 | check_returncode: bool = True, |
136 | | - callback: Union[Callable[[str, datetime.datetime], None], None] = None, |
| 144 | + callback: Optional[ProgressCallbackProtocol] = None, |
137 | 145 | ): |
138 | 146 | """Run 'cmd' in a shell and return the combined contents of stdout and |
139 | 147 | stderr (Blocking). Throws an exception if the command exits non-zero. |
@@ -161,7 +169,7 @@ def run( |
161 | 169 | Indicate whether a `libvcs.exc.CommandError` should be raised if return code is |
162 | 170 | different from 0. |
163 | 171 |
|
164 | | - callback : callable |
| 172 | + callback : ProgressCallbackProtocol |
165 | 173 | callback to return output as a command executes, accepts a function signature |
166 | 174 | of `(output, timestamp)`. Example usage:: |
167 | 175 |
|
|
0 commit comments