14
14
import logging
15
15
import subprocess
16
16
import sys
17
+ import typing as t
17
18
from collections .abc import Iterable , Mapping , MutableMapping , Sequence
18
- from typing import IO , TYPE_CHECKING , Any , AnyStr , Callable , Optional , Protocol , Union
19
19
20
20
from libvcs import exc
21
21
from libvcs ._internal .types import StrOrBytesPath
@@ -35,7 +35,7 @@ def console_to_str(s: bytes) -> str:
35
35
return str (s )
36
36
37
37
38
- if TYPE_CHECKING :
38
+ if t . TYPE_CHECKING :
39
39
_LoggerAdapter = logging .LoggerAdapter [logging .Logger ]
40
40
from typing_extensions import TypeAlias
41
41
else :
@@ -58,7 +58,9 @@ class CmdLoggingAdapter(_LoggerAdapter):
58
58
directory basename, name of repo, hint, etc. e.g. 'django'
59
59
"""
60
60
61
- def __init__ (self , bin_name : str , keyword : str , * args : Any , ** kwargs : Any ) -> None :
61
+ def __init__ (
62
+ self , bin_name : str , keyword : str , * args : t .Any , ** kwargs : t .Any
63
+ ) -> None :
62
64
#: bin_name
63
65
self .bin_name = bin_name
64
66
#: directory basename, name of repository, hint, etc.
@@ -69,8 +71,8 @@ def __init__(self, bin_name: str, keyword: str, *args: Any, **kwargs: Any) -> No
69
71
def process (
70
72
self ,
71
73
msg : str ,
72
- kwargs : MutableMapping [str , Any ],
73
- ) -> tuple [Any , MutableMapping [str , Any ]]:
74
+ kwargs : MutableMapping [str , t . Any ],
75
+ ) -> tuple [t . Any , MutableMapping [str , t . Any ]]:
74
76
"""Add additional context information for loggers."""
75
77
prefixed_dict = {}
76
78
prefixed_dict ["bin_name" ] = self .bin_name
@@ -81,24 +83,24 @@ def process(
81
83
return msg , kwargs
82
84
83
85
84
- class ProgressCallbackProtocol (Protocol ):
86
+ class ProgressCallbackProtocol (t . Protocol ):
85
87
"""Callback to report subprocess communication."""
86
88
87
- def __call__ (self , output : AnyStr , timestamp : datetime .datetime ) -> None :
89
+ def __call__ (self , output : t . AnyStr , timestamp : datetime .datetime ) -> None :
88
90
"""Process progress for subprocess communication."""
89
91
...
90
92
91
93
92
94
if sys .platform == "win32" :
93
95
_ENV : TypeAlias = Mapping [str , str ]
94
96
else :
95
- _ENV : TypeAlias = Union [
97
+ _ENV : TypeAlias = t . Union [
96
98
Mapping [bytes , StrOrBytesPath ],
97
99
Mapping [str , StrOrBytesPath ],
98
100
]
99
101
100
- _CMD = Union [StrOrBytesPath , Sequence [StrOrBytesPath ]]
101
- _FILE : TypeAlias = Optional [Union [int , IO [Any ]]]
102
+ _CMD = t . Union [StrOrBytesPath , Sequence [StrOrBytesPath ]]
103
+ _FILE : TypeAlias = t . Optional [t . Union [int , t . IO [t . Any ]]]
102
104
103
105
104
106
def run (
@@ -108,17 +110,17 @@ def run(
108
110
stdin : _FILE | None = None ,
109
111
stdout : _FILE | None = None ,
110
112
stderr : _FILE | None = None ,
111
- preexec_fn : Callable [[], Any ] | None = None ,
113
+ preexec_fn : t . Callable [[], t . Any ] | None = None ,
112
114
close_fds : bool = True ,
113
115
shell : bool = False ,
114
116
cwd : StrOrBytesPath | None = None ,
115
117
env : _ENV | None = None ,
116
118
universal_newlines : bool = False ,
117
- startupinfo : Any | None = None ,
119
+ startupinfo : t . Any | None = None ,
118
120
creationflags : int = 0 ,
119
121
restore_signals : bool = True ,
120
122
start_new_session : bool = False ,
121
- pass_fds : Any = (),
123
+ pass_fds : t . Any = (),
122
124
* ,
123
125
text : bool | None = None ,
124
126
encoding : str | None = None ,
@@ -205,7 +207,7 @@ def progress_cb(output, timestamp):
205
207
line = None
206
208
if log_in_real_time and callback is None :
207
209
208
- def progress_cb (output : AnyStr , timestamp : datetime .datetime ) -> None :
210
+ def progress_cb (output : t . AnyStr , timestamp : datetime .datetime ) -> None :
209
211
sys .stdout .write (str (output ))
210
212
sys .stdout .flush ()
211
213
0 commit comments