mirror of
https://github.com/davidhalter/typeshed.git
synced 2025-12-06 20:24:30 +08:00
87 lines
3.1 KiB
Python
87 lines
3.1 KiB
Python
from _typeshed import structseq
|
|
from typing import Any, Final, Literal, SupportsIndex, final
|
|
from typing_extensions import Buffer, Self
|
|
|
|
class ChannelError(RuntimeError): ...
|
|
class ChannelClosedError(ChannelError): ...
|
|
class ChannelEmptyError(ChannelError): ...
|
|
class ChannelNotEmptyError(ChannelError): ...
|
|
class ChannelNotFoundError(ChannelError): ...
|
|
|
|
# Mark as final, since instantiating ChannelID is not supported.
|
|
@final
|
|
class ChannelID:
|
|
@property
|
|
def end(self) -> Literal["send", "recv", "both"]: ...
|
|
@property
|
|
def send(self) -> Self: ...
|
|
@property
|
|
def recv(self) -> Self: ...
|
|
def __eq__(self, other: object) -> bool: ...
|
|
def __ge__(self, other: ChannelID) -> bool: ...
|
|
def __gt__(self, other: ChannelID) -> bool: ...
|
|
def __hash__(self) -> int: ...
|
|
def __index__(self) -> int: ...
|
|
def __int__(self) -> int: ...
|
|
def __le__(self, other: ChannelID) -> bool: ...
|
|
def __lt__(self, other: ChannelID) -> bool: ...
|
|
def __ne__(self, other: object) -> bool: ...
|
|
|
|
@final
|
|
class ChannelInfo(structseq[int], tuple[bool, bool, bool, int, int, int, int, int]):
|
|
__match_args__: Final = (
|
|
"open",
|
|
"closing",
|
|
"closed",
|
|
"count",
|
|
"num_interp_send",
|
|
"num_interp_send_released",
|
|
"num_interp_recv",
|
|
"num_interp_recv_released",
|
|
)
|
|
@property
|
|
def open(self) -> bool: ...
|
|
@property
|
|
def closing(self) -> bool: ...
|
|
@property
|
|
def closed(self) -> bool: ...
|
|
@property
|
|
def count(self) -> int: ... # type: ignore[override]
|
|
@property
|
|
def num_interp_send(self) -> int: ...
|
|
@property
|
|
def num_interp_send_released(self) -> int: ...
|
|
@property
|
|
def num_interp_recv(self) -> int: ...
|
|
@property
|
|
def num_interp_recv_released(self) -> int: ...
|
|
@property
|
|
def num_interp_both(self) -> int: ...
|
|
@property
|
|
def num_interp_both_recv_released(self) -> int: ...
|
|
@property
|
|
def num_interp_both_send_released(self) -> int: ...
|
|
@property
|
|
def num_interp_both_released(self) -> int: ...
|
|
@property
|
|
def recv_associated(self) -> bool: ...
|
|
@property
|
|
def recv_released(self) -> bool: ...
|
|
@property
|
|
def send_associated(self) -> bool: ...
|
|
@property
|
|
def send_released(self) -> bool: ...
|
|
|
|
def create(unboundop: Literal[1, 2, 3]) -> ChannelID: ...
|
|
def destroy(cid: SupportsIndex) -> None: ...
|
|
def list_all() -> list[ChannelID]: ...
|
|
def list_interpreters(cid: SupportsIndex, *, send: bool) -> list[int]: ...
|
|
def send(cid: SupportsIndex, obj: object, *, blocking: bool = True, timeout: float | None = None) -> None: ...
|
|
def send_buffer(cid: SupportsIndex, obj: Buffer, *, blocking: bool = True, timeout: float | None = None) -> None: ...
|
|
def recv(cid: SupportsIndex, default: object = ...) -> tuple[Any, Literal[1, 2, 3]]: ...
|
|
def close(cid: SupportsIndex, *, send: bool = False, recv: bool = False) -> None: ...
|
|
def get_count(cid: SupportsIndex) -> int: ...
|
|
def get_info(cid: SupportsIndex) -> ChannelInfo: ...
|
|
def get_channel_defaults(cid: SupportsIndex) -> Literal[1, 2, 3]: ...
|
|
def release(cid: SupportsIndex, *, send: bool = False, recv: bool = False, force: bool = False) -> None: ...
|