13
13
import sys
14
14
import time
15
15
import typing as t
16
+ from ctypes import Array
16
17
from ctypes import byref
17
18
from ctypes import c_char
18
19
from ctypes import c_char_p
67
68
EOF = b"\x1a "
68
69
MAX_BYTES_WRITTEN = 32767
69
70
71
+ if t .TYPE_CHECKING :
72
+ try :
73
+ # Using `typing_extensions.Buffer` instead of `collections.abc`
74
+ # on Windows for some reason does not have `Sized` implemented.
75
+ from collections .abc import Buffer # type: ignore
76
+ except ImportError :
77
+ from typing_extensions import Buffer
78
+
70
79
try :
71
80
from ctypes import pythonapi
72
81
except ImportError :
@@ -93,32 +102,32 @@ class Py_buffer(Structure):
93
102
PyObject_GetBuffer = pythonapi .PyObject_GetBuffer
94
103
PyBuffer_Release = pythonapi .PyBuffer_Release
95
104
96
- def get_buffer (obj , writable = False ):
105
+ def get_buffer (obj : Buffer , writable : bool = False ) -> Array [ c_char ] :
97
106
buf = Py_buffer ()
98
- flags = PyBUF_WRITABLE if writable else PyBUF_SIMPLE
107
+ flags : int = PyBUF_WRITABLE if writable else PyBUF_SIMPLE
99
108
PyObject_GetBuffer (py_object (obj ), byref (buf ), flags )
100
109
101
110
try :
102
- buffer_type = c_char * buf .len
111
+ buffer_type : Array [ c_char ] = c_char * buf .len
103
112
return buffer_type .from_address (buf .buf )
104
113
finally :
105
114
PyBuffer_Release (byref (buf ))
106
115
107
116
108
117
class _WindowsConsoleRawIOBase (io .RawIOBase ):
109
- def __init__ (self , handle ) :
118
+ def __init__ (self , handle : int | None ) -> None :
110
119
self .handle = handle
111
120
112
- def isatty (self ):
121
+ def isatty (self ) -> t . Literal [ True ] :
113
122
super ().isatty ()
114
123
return True
115
124
116
125
117
126
class _WindowsConsoleReader (_WindowsConsoleRawIOBase ):
118
- def readable (self ):
127
+ def readable (self ) -> t . Literal [ True ] :
119
128
return True
120
129
121
- def readinto (self , b ) :
130
+ def readinto (self , b : Buffer ) -> int :
122
131
bytes_to_be_read = len (b )
123
132
if not bytes_to_be_read :
124
133
return 0
@@ -150,18 +159,18 @@ def readinto(self, b):
150
159
151
160
152
161
class _WindowsConsoleWriter (_WindowsConsoleRawIOBase ):
153
- def writable (self ):
162
+ def writable (self ) -> t . Literal [ True ] :
154
163
return True
155
164
156
165
@staticmethod
157
- def _get_error_message (errno ) :
166
+ def _get_error_message (errno : int ) -> str :
158
167
if errno == ERROR_SUCCESS :
159
168
return "ERROR_SUCCESS"
160
169
elif errno == ERROR_NOT_ENOUGH_MEMORY :
161
170
return "ERROR_NOT_ENOUGH_MEMORY"
162
171
return f"Windows error { errno } "
163
172
164
- def write (self , b ) :
173
+ def write (self , b : Buffer ) -> int :
165
174
bytes_to_be_written = len (b )
166
175
buf = get_buffer (b )
167
176
code_units_to_be_written = min (bytes_to_be_written , MAX_BYTES_WRITTEN ) // 2
@@ -209,7 +218,7 @@ def __getattr__(self, name: str) -> t.Any:
209
218
def isatty (self ) -> bool :
210
219
return self .buffer .isatty ()
211
220
212
- def __repr__ (self ):
221
+ def __repr__ (self ) -> str :
213
222
return f"<ConsoleStream name={ self .name !r} encoding={ self .encoding !r} >"
214
223
215
224
@@ -267,16 +276,20 @@ def _get_windows_console_stream(
267
276
f : t .TextIO , encoding : str | None , errors : str | None
268
277
) -> t .TextIO | None :
269
278
if (
270
- get_buffer is not None
271
- and encoding in {"utf-16-le" , None }
272
- and errors in {"strict" , None }
273
- and _is_console (f )
279
+ get_buffer is None
280
+ or encoding not in {"utf-16-le" , None }
281
+ or errors not in {"strict" , None }
282
+ or not _is_console (f )
274
283
):
275
- func = _stream_factories .get (f .fileno ())
276
- if func is not None :
277
- b = getattr (f , "buffer" , None )
284
+ return None
285
+
286
+ func = _stream_factories .get (f .fileno ())
287
+ if func is None :
288
+ return None
289
+
290
+ b = getattr (f , "buffer" , None )
278
291
279
- if b is None :
280
- return None
292
+ if b is None :
293
+ return None
281
294
282
- return func (b )
295
+ return func (b )
0 commit comments