io.websockets
ServerConnection
class ServerConnection(Protocol)
发送
def send(message: Union[Data, Iterable[Data]]) -> None
向客户端发送消息。
参数:
message
Union[Data, Iterable[Data]] - 要发送的消息。
recv
def recv(timeout: Optional[float] = None) -> Data
从客户端接收消息。
参数:
timeout
Optional[float], optional - 接收操作的超时时间。默认为 None。
返回:
Data
- 从客户端接收到的消息。
关闭
def close() -> None
关闭连接。
WebSocket服务器
class WebSocketServer(Protocol)
serve_forever
def serve_forever() -> None
永远运行服务器。
关闭
def shutdown() -> None
关闭服务器。
__enter__
def __enter__() -> "WebSocketServer"
进入服务器上下文。
__exit__
def __exit__(exc_type: Any, exc_value: Any, traceback: Any) -> None
退出服务器上下文。
IOWebsockets
class IOWebsockets(IOStream)
一个websocket输入/输出流。
__init__
def __init__(websocket: ServerConnection) -> None
初始化WebSocket输入/输出流。
参数:
websocket
ServerConnection - websocket 服务器。
引发:
ImportError
- 如果 websockets 模块不可用。
run_server_in_thread
@staticmethod
@contextmanager
def run_server_in_thread(*,
host: str = "127.0.0.1",
port: int = 8765,
on_connect: Callable[["IOWebsockets"], None],
ssl_context: Optional[ssl.SSLContext] = None,
**kwargs: Any) -> Iterator[str]
用于创建websocket输入/输出流的工厂函数。
参数:
host
字符串, 可选 - 服务器绑定的主机。默认为 "127.0.0.1"。port
int, 可选 - 服务器绑定的端口。默认为8765。on_connect
Callable[[IOWebsockets], None] - 客户端连接时执行的函数。通常用于创建代理并启动聊天。ssl_context
Optional[ssl.SSLContext], optional - 用于安全连接的SSL上下文。默认为None。kwargs
Any - 传递给websocket服务器的额外关键字参数。
产出:
str
- WebSocket服务器的URI。
websocket
@property
def websocket() -> "ServerConnection"
WebSocket 服务器的 URI。
打印
def print(*objects: Any,
sep: str = " ",
end: str = "\n",
flush: bool = False) -> None
将数据打印到输出流。
参数:
objects
任何 - 要打印的数据。sep
str, 可选 - 对象之间的分隔符。默认为 " ".end
str, 可选 - 输出的结尾。默认值为 " ".flush
布尔值,可选 - 是否刷新输出。默认为 False。
输入
def input(prompt: str = "", *, password: bool = False) -> str
从输入流中读取一行。
参数:
prompt
str, 可选 - 要显示的提示信息。默认为 ""。password
布尔值, 可选 - 是否读取密码。默认为 False。
返回:
str
- 从输入流中读取的行。