mirror of
https://github.com/davidhalter/typeshed.git
synced 2025-12-09 05:24:52 +08:00
Use I/O protocols in codecs.pyi (#7849)
This commit is contained in:
@@ -2,8 +2,8 @@ import sys
|
||||
import types
|
||||
from _typeshed import Self
|
||||
from abc import abstractmethod
|
||||
from collections.abc import Callable, Generator, Iterable, Iterator
|
||||
from typing import IO, Any, BinaryIO, Protocol, TextIO, overload
|
||||
from collections.abc import Callable, Generator, Iterable
|
||||
from typing import Any, BinaryIO, Protocol, TextIO, overload
|
||||
from typing_extensions import Literal, TypeAlias
|
||||
|
||||
__all__ = [
|
||||
@@ -58,6 +58,18 @@ BOM32_LE: Literal[b"\xff\xfe"]
|
||||
BOM64_BE: Literal[b"\x00\x00\xfe\xff"]
|
||||
BOM64_LE: Literal[b"\xff\xfe\x00\x00"]
|
||||
|
||||
class _WritableStream(Protocol):
|
||||
def write(self, __data: bytes) -> object: ...
|
||||
def seek(self, __offset: int, __whence: int) -> object: ...
|
||||
def close(self) -> object: ...
|
||||
|
||||
class _ReadableStream(Protocol):
|
||||
def read(self, __size: int = ...) -> bytes: ...
|
||||
def seek(self, __offset: int, __whence: int) -> object: ...
|
||||
def close(self) -> object: ...
|
||||
|
||||
class _Stream(_WritableStream, _ReadableStream, Protocol): ...
|
||||
|
||||
# TODO: this only satisfies the most common interface, where
|
||||
# bytes is the raw form and str is the cooked form.
|
||||
# In the long run, both should become template parameters maybe?
|
||||
@@ -71,10 +83,10 @@ class _Decoder(Protocol):
|
||||
def __call__(self, input: bytes, errors: str = ...) -> tuple[str, int]: ... # signature of Codec().decode
|
||||
|
||||
class _StreamReader(Protocol):
|
||||
def __call__(self, stream: IO[bytes], errors: str = ...) -> StreamReader: ...
|
||||
def __call__(self, stream: _ReadableStream, errors: str = ...) -> StreamReader: ...
|
||||
|
||||
class _StreamWriter(Protocol):
|
||||
def __call__(self, stream: IO[bytes], errors: str = ...) -> StreamWriter: ...
|
||||
def __call__(self, stream: _WritableStream, errors: str = ...) -> StreamWriter: ...
|
||||
|
||||
class _IncrementalEncoder(Protocol):
|
||||
def __call__(self, errors: str = ...) -> IncrementalEncoder: ...
|
||||
@@ -162,7 +174,7 @@ def register(__search_function: Callable[[str], CodecInfo | None]) -> None: ...
|
||||
def open(
|
||||
filename: str, mode: str = ..., encoding: str | None = ..., errors: str = ..., buffering: int = ...
|
||||
) -> StreamReaderWriter: ...
|
||||
def EncodedFile(file: IO[bytes], data_encoding: str, file_encoding: str | None = ..., errors: str = ...) -> StreamRecoder: ...
|
||||
def EncodedFile(file: _Stream, data_encoding: str, file_encoding: str | None = ..., errors: str = ...) -> StreamRecoder: ...
|
||||
def iterencode(iterator: Iterable[str], encoding: str, errors: str = ...) -> Generator[bytes, None, None]: ...
|
||||
def iterdecode(iterator: Iterable[bytes], encoding: str, errors: str = ...) -> Generator[str, None, None]: ...
|
||||
|
||||
@@ -235,8 +247,9 @@ class BufferedIncrementalDecoder(IncrementalDecoder):
|
||||
# TODO: it is not possible to specify the requirement that all other
|
||||
# attributes and methods are passed-through from the stream.
|
||||
class StreamWriter(Codec):
|
||||
stream: _WritableStream
|
||||
errors: str
|
||||
def __init__(self, stream: IO[bytes], errors: str = ...) -> None: ...
|
||||
def __init__(self, stream: _WritableStream, errors: str = ...) -> None: ...
|
||||
def write(self, object: str) -> None: ...
|
||||
def writelines(self, list: Iterable[str]) -> None: ...
|
||||
def reset(self) -> None: ...
|
||||
@@ -244,9 +257,10 @@ class StreamWriter(Codec):
|
||||
def __exit__(self, type: type[BaseException] | None, value: BaseException | None, tb: types.TracebackType | None) -> None: ...
|
||||
def __getattr__(self, name: str, getattr: Callable[[str], Any] = ...) -> Any: ...
|
||||
|
||||
class StreamReader(Codec, Iterator[str]):
|
||||
class StreamReader(Codec):
|
||||
stream: _ReadableStream
|
||||
errors: str
|
||||
def __init__(self, stream: IO[bytes], errors: str = ...) -> None: ...
|
||||
def __init__(self, stream: _ReadableStream, errors: str = ...) -> None: ...
|
||||
def read(self, size: int = ..., chars: int = ..., firstline: bool = ...) -> str: ...
|
||||
def readline(self, size: int | None = ..., keepends: bool = ...) -> str: ...
|
||||
def readlines(self, sizehint: int | None = ..., keepends: bool = ...) -> list[str]: ...
|
||||
@@ -260,7 +274,8 @@ class StreamReader(Codec, Iterator[str]):
|
||||
# Doesn't actually inherit from TextIO, but wraps a BinaryIO to provide text reading and writing
|
||||
# and delegates attributes to the underlying binary stream with __getattr__.
|
||||
class StreamReaderWriter(TextIO):
|
||||
def __init__(self, stream: IO[bytes], Reader: _StreamReader, Writer: _StreamWriter, errors: str = ...) -> None: ...
|
||||
stream: _Stream
|
||||
def __init__(self, stream: _Stream, Reader: _StreamReader, Writer: _StreamWriter, errors: str = ...) -> None: ...
|
||||
def read(self, size: int = ...) -> str: ...
|
||||
def readline(self, size: int | None = ...) -> str: ...
|
||||
def readlines(self, sizehint: int | None = ...) -> list[str]: ...
|
||||
@@ -287,13 +302,7 @@ class StreamReaderWriter(TextIO):
|
||||
|
||||
class StreamRecoder(BinaryIO):
|
||||
def __init__(
|
||||
self,
|
||||
stream: IO[bytes],
|
||||
encode: _Encoder,
|
||||
decode: _Decoder,
|
||||
Reader: _StreamReader,
|
||||
Writer: _StreamWriter,
|
||||
errors: str = ...,
|
||||
self, stream: _Stream, encode: _Encoder, decode: _Decoder, Reader: _StreamReader, Writer: _StreamWriter, errors: str = ...
|
||||
) -> None: ...
|
||||
def read(self, size: int = ...) -> bytes: ...
|
||||
def readline(self, size: int | None = ...) -> bytes: ...
|
||||
|
||||
Reference in New Issue
Block a user