Format: application/octet-stream
1 import sys
2 import types
3 from abc import abstractmethod
4 from typing import (
5 IO,
6 Any,
7 BinaryIO,
8 Callable,
9 Generator,
10 Iterable,
11 Iterator,
12 List,
13 Optional,
14 Protocol,
15 Text,
16 TextIO,
17 Tuple,
18 Type,
19 TypeVar,
20 Union,
21 overload,
22 )
23 from typing_extensions import Literal
24
25 # TODO: this only satisfies the most common interface, where
26 # bytes (py2 str) is the raw form and str (py2 unicode) is the cooked form.
27 # In the long run, both should become template parameters maybe?
28 # There *are* bytes->bytes and str->str encodings in the standard library.
29 # They are much more common in Python 2 than in Python 3.
30
31 _Decoded = Text
32 _Encoded = bytes
33
34 class _Encoder(Protocol):
35 def __call__(self, input: _Decoded, errors: str = ...) -> Tuple[_Encoded, int]: ... # signature of Codec().encode
36
37 class _Decoder(Protocol):
38 def __call__(self, input: _Encoded, errors: str = ...) -> Tuple[_Decoded, int]: ... # signature of Codec().decode
39
40 class _StreamReader(Protocol):
41 def __call__(self, stream: IO[_Encoded], errors: str = ...) -> StreamReader: ...
42
43 class _StreamWriter(Protocol):
44 def __call__(self, stream: IO[_Encoded], errors: str = ...) -> StreamWriter: ...
45
46 class _IncrementalEncoder(Protocol):
47 def __call__(self, errors: str = ...) -> IncrementalEncoder: ...
48
49 class _IncrementalDecoder(Protocol):
50 def __call__(self, errors: str = ...) -> IncrementalDecoder: ...
51
52 # The type ignore on `encode` and `decode` is to avoid issues with overlapping overloads, for more details, see #300
53 # mypy and pytype disagree about where the type ignore can and cannot go, so alias the long type
54 _BytesToBytesEncodingT = Literal[
55 "base64",
56 "base_64",
57 "base64_codec",
58 "bz2",
59 "bz2_codec",
60 "hex",
61 "hex_codec",
62 "quopri",
63 "quotedprintable",
64 "quoted_printable",
65 "quopri_codec",
66 "uu",
67 "uu_codec",
68 "zip",
69 "zlib",
70 "zlib_codec",
71 ]
72 @overload
73 def encode(obj: bytes, encoding: _BytesToBytesEncodingT, errors: str = ...) -> bytes: ...
74 @overload
75 def encode(obj: str, encoding: Literal["rot13", "rot_13"] = ..., errors: str = ...) -> str: ... # type: ignore
76 @overload
77 def encode(obj: _Decoded, encoding: str = ..., errors: str = ...) -> _Encoded: ...
78 @overload
79 def decode(obj: bytes, encoding: _BytesToBytesEncodingT, errors: str = ...) -> bytes: ... # type: ignore
80 @overload
81 def decode(obj: str, encoding: Literal["rot13", "rot_13"] = ..., errors: str = ...) -> Text: ...
82 @overload
83 def decode(obj: _Encoded, encoding: str = ..., errors: str = ...) -> _Decoded: ...
84 def lookup(__encoding: str) -> CodecInfo: ...
85 def utf_16_be_decode(__obj: _Encoded, __errors: str = ..., __final: bool = ...) -> Tuple[_Decoded, int]: ... # undocumented
86 def utf_16_be_encode(__obj: _Decoded, __errors: str = ...) -> Tuple[_Encoded, int]: ... # undocumented
87
88 class CodecInfo(Tuple[_Encoder, _Decoder, _StreamReader, _StreamWriter]):
89 @property
90 def encode(self) -> _Encoder: ...
91 @property
92 def decode(self) -> _Decoder: ...
93 @property
94 def streamreader(self) -> _StreamReader: ...
95 @property
96 def streamwriter(self) -> _StreamWriter: ...
97 @property
98 def incrementalencoder(self) -> _IncrementalEncoder: ...
99 @property
100 def incrementaldecoder(self) -> _IncrementalDecoder: ...
101 name: str
102 def __new__(
103 cls,
104 encode: _Encoder,
105 decode: _Decoder,
106 streamreader: Optional[_StreamReader] = ...,
107 streamwriter: Optional[_StreamWriter] = ...,
108 incrementalencoder: Optional[_IncrementalEncoder] = ...,
109 incrementaldecoder: Optional[_IncrementalDecoder] = ...,
110 name: Optional[str] = ...,
111 *,
112 _is_text_encoding: Optional[bool] = ...,
113 ) -> CodecInfo: ...
114
115 def getencoder(encoding: str) -> _Encoder: ...
116 def getdecoder(encoding: str) -> _Decoder: ...
117 def getincrementalencoder(encoding: str) -> _IncrementalEncoder: ...
118 def getincrementaldecoder(encoding: str) -> _IncrementalDecoder: ...
119 def getreader(encoding: str) -> _StreamReader: ...
120 def getwriter(encoding: str) -> _StreamWriter: ...
121 def register(__search_function: Callable[[str], Optional[CodecInfo]]) -> None: ...
122 def open(
123 filename: str, mode: str = ..., encoding: Optional[str] = ..., errors: str = ..., buffering: int = ...
124 ) -> StreamReaderWriter: ...
125 def EncodedFile(
126 file: IO[_Encoded], data_encoding: str, file_encoding: Optional[str] = ..., errors: str = ...
127 ) -> StreamRecoder: ...
128 def iterencode(iterator: Iterable[_Decoded], encoding: str, errors: str = ...) -> Generator[_Encoded, None, None]: ...
129 def iterdecode(iterator: Iterable[_Encoded], encoding: str, errors: str = ...) -> Generator[_Decoded, None, None]: ...
130
131 BOM: bytes
132 BOM_BE: bytes
133 BOM_LE: bytes
134 BOM_UTF8: bytes
135 BOM_UTF16: bytes
136 BOM_UTF16_BE: bytes
137 BOM_UTF16_LE: bytes
138 BOM_UTF32: bytes
139 BOM_UTF32_BE: bytes
140 BOM_UTF32_LE: bytes
141
142 # It is expected that different actions be taken depending on which of the
143 # three subclasses of `UnicodeError` is actually ...ed. However, the Union
144 # is still needed for at least one of the cases.
145 def register_error(__errors: str, __handler: Callable[[UnicodeError], Tuple[Union[str, bytes], int]]) -> None: ...
146 def lookup_error(__name: str) -> Callable[[UnicodeError], Tuple[Union[str, bytes], int]]: ...
147 def strict_errors(exception: UnicodeError) -> Tuple[Union[str, bytes], int]: ...
148 def replace_errors(exception: UnicodeError) -> Tuple[Union[str, bytes], int]: ...
149 def ignore_errors(exception: UnicodeError) -> Tuple[Union[str, bytes], int]: ...
150 def xmlcharrefreplace_errors(exception: UnicodeError) -> Tuple[Union[str, bytes], int]: ...
151 def backslashreplace_errors(exception: UnicodeError) -> Tuple[Union[str, bytes], int]: ...
152
153 class Codec:
154 # These are sort of @abstractmethod but sort of not.
155 # The StreamReader and StreamWriter subclasses only implement one.
156 def encode(self, input: _Decoded, errors: str = ...) -> Tuple[_Encoded, int]: ...
157 def decode(self, input: _Encoded, errors: str = ...) -> Tuple[_Decoded, int]: ...
158
159 class IncrementalEncoder:
160 errors: str
161 def __init__(self, errors: str = ...) -> None: ...
162 @abstractmethod
163 def encode(self, object: _Decoded, final: bool = ...) -> _Encoded: ...
164 def reset(self) -> None: ...
165 # documentation says int but str is needed for the subclass.
166 def getstate(self) -> Union[int, _Decoded]: ...
167 def setstate(self, state: Union[int, _Decoded]) -> None: ...
168
169 class IncrementalDecoder:
170 errors: str
171 def __init__(self, errors: str = ...) -> None: ...
172 @abstractmethod
173 def decode(self, input: _Encoded, final: bool = ...) -> _Decoded: ...
174 def reset(self) -> None: ...
175 def getstate(self) -> Tuple[_Encoded, int]: ...
176 def setstate(self, state: Tuple[_Encoded, int]) -> None: ...
177
178 # These are not documented but used in encodings/*.py implementations.
179 class BufferedIncrementalEncoder(IncrementalEncoder):
180 buffer: str
181 def __init__(self, errors: str = ...) -> None: ...
182 @abstractmethod
183 def _buffer_encode(self, input: _Decoded, errors: str, final: bool) -> _Encoded: ...
184 def encode(self, input: _Decoded, final: bool = ...) -> _Encoded: ...
185
186 class BufferedIncrementalDecoder(IncrementalDecoder):
187 buffer: bytes
188 def __init__(self, errors: str = ...) -> None: ...
189 @abstractmethod
190 def _buffer_decode(self, input: _Encoded, errors: str, final: bool) -> Tuple[_Decoded, int]: ...
191 def decode(self, object: _Encoded, final: bool = ...) -> _Decoded: ...
192
193 _SW = TypeVar("_SW", bound=StreamWriter)
194
195 # TODO: it is not possible to specify the requirement that all other
196 # attributes and methods are passed-through from the stream.
197 class StreamWriter(Codec):
198 errors: str
199 def __init__(self, stream: IO[_Encoded], errors: str = ...) -> None: ...
200 def write(self, obj: _Decoded) -> None: ...
201 def writelines(self, list: Iterable[_Decoded]) -> None: ...
202 def reset(self) -> None: ...
203 def __enter__(self: _SW) -> _SW: ...
204 def __exit__(
205 self, typ: Optional[Type[BaseException]], exc: Optional[BaseException], tb: Optional[types.TracebackType]
206 ) -> None: ...
207 def __getattr__(self, name: str) -> Any: ...
208
209 _SR = TypeVar("_SR", bound=StreamReader)
210
211 class StreamReader(Codec):
212 errors: str
213 def __init__(self, stream: IO[_Encoded], errors: str = ...) -> None: ...
214 def read(self, size: int = ..., chars: int = ..., firstline: bool = ...) -> _Decoded: ...
215 def readline(self, size: int = ..., keepends: bool = ...) -> _Decoded: ...
216 def readlines(self, sizehint: int = ..., keepends: bool = ...) -> List[_Decoded]: ...
217 def reset(self) -> None: ...
218 def __enter__(self: _SR) -> _SR: ...
219 def __exit__(
220 self, typ: Optional[Type[BaseException]], exc: Optional[BaseException], tb: Optional[types.TracebackType]
221 ) -> None: ...
222 def __iter__(self) -> Iterator[_Decoded]: ...
223 def __getattr__(self, name: str) -> Any: ...
224
225 _T = TypeVar("_T", bound=StreamReaderWriter)
226
227 # Doesn't actually inherit from TextIO, but wraps a BinaryIO to provide text reading and writing
228 # and delegates attributes to the underlying binary stream with __getattr__.
229 class StreamReaderWriter(TextIO):
230 def __init__(self, stream: IO[_Encoded], Reader: _StreamReader, Writer: _StreamWriter, errors: str = ...) -> None: ...
231 def read(self, size: int = ...) -> _Decoded: ...
232 def readline(self, size: Optional[int] = ...) -> _Decoded: ...
233 def readlines(self, sizehint: Optional[int] = ...) -> List[_Decoded]: ...
234 if sys.version_info >= (3,):
235 def __next__(self) -> Text: ...
236 else:
237 def next(self) -> Text: ...
238 def __iter__(self: _T) -> _T: ...
239 # This actually returns None, but that's incompatible with the supertype
240 def write(self, data: _Decoded) -> int: ...
241 def writelines(self, list: Iterable[_Decoded]) -> None: ...
242 def reset(self) -> None: ...
243 # Same as write()
244 def seek(self, offset: int, whence: int = ...) -> int: ...
245 def __enter__(self: _T) -> _T: ...
246 def __exit__(
247 self, typ: Optional[Type[BaseException]], exc: Optional[BaseException], tb: Optional[types.TracebackType]
248 ) -> None: ...
249 def __getattr__(self, name: str) -> Any: ...
250 # These methods don't actually exist directly, but they are needed to satisfy the TextIO
251 # interface. At runtime, they are delegated through __getattr__.
252 def close(self) -> None: ...
253 def fileno(self) -> int: ...
254 def flush(self) -> None: ...
255 def isatty(self) -> bool: ...
256 def readable(self) -> bool: ...
257 def truncate(self, size: Optional[int] = ...) -> int: ...
258 def seekable(self) -> bool: ...
259 def tell(self) -> int: ...
260 def writable(self) -> bool: ...
261
262 _SRT = TypeVar("_SRT", bound=StreamRecoder)
263
264 class StreamRecoder(BinaryIO):
265 def __init__(
266 self,
267 stream: IO[_Encoded],
268 encode: _Encoder,
269 decode: _Decoder,
270 Reader: _StreamReader,
271 Writer: _StreamWriter,
272 errors: str = ...,
273 ) -> None: ...
274 def read(self, size: int = ...) -> bytes: ...
275 def readline(self, size: Optional[int] = ...) -> bytes: ...
276 def readlines(self, sizehint: Optional[int] = ...) -> List[bytes]: ...
277 if sys.version_info >= (3,):
278 def __next__(self) -> bytes: ...
279 else:
280 def next(self) -> bytes: ...
281 def __iter__(self: _SRT) -> _SRT: ...
282 def write(self, data: bytes) -> int: ...
283 def writelines(self, list: Iterable[bytes]) -> int: ... # type: ignore # it's supposed to return None
284 def reset(self) -> None: ...
285 def __getattr__(self, name: str) -> Any: ...
286 def __enter__(self: _SRT) -> _SRT: ...
287 def __exit__(
288 self, type: Optional[Type[BaseException]], value: Optional[BaseException], tb: Optional[types.TracebackType]
289 ) -> None: ...
290 # These methods don't actually exist directly, but they are needed to satisfy the BinaryIO
291 # interface. At runtime, they are delegated through __getattr__.
292 def seek(self, offset: int, whence: int = ...) -> int: ...
293 def close(self) -> None: ...
294 def fileno(self) -> int: ...
295 def flush(self) -> None: ...
296 def isatty(self) -> bool: ...
297 def readable(self) -> bool: ...
298 def truncate(self, size: Optional[int] = ...) -> int: ...
299 def seekable(self) -> bool: ...
300 def tell(self) -> int: ...
301 def writable(self) -> bool: ...
302