1 from threading import Event
2 from typing import Generic, Text, TypeVar
4 _T = TypeVar("_T", Text, bytes)
6 class PipeTimeout(IOError): ...
8 class BufferedPipe(Generic[_T]):
9 def __init__(self) -> None: ...
10 def set_event(self, event: Event) -> None: ...
11 def feed(self, data: _T) -> None: ...
12 def read_ready(self) -> bool: ...
13 def read(self, nbytes: int, timeout: float | None = ...) -> _T: ...
14 def empty(self) -> _T: ...
15 def close(self) -> None: ...
16 def __len__(self) -> int: ...