跳到主要内容

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 - 从输入流中读取的行。