websocket - PythonIDE Docs

原生 WebSocket 客户端（URLSessionWebSocketTask）：连接、收发消息、事件回调。

边界：客户端连接 ws:// / wss:// 服务器；不含 WebSocket 服务端。阻塞式 receive() 会等待消息；AppUI 场景推荐 on_message 回调。

#模块概览

项	说明
导入	`import websocket`
适合做什么	实时聊天、行情推送、协作白板
调用时机	`connect` 放按钮回调；长连接用 `on_message`
推荐顺序	`connect(url)` → `send` → `receive` 或设回调 → `close()`
状态	`ws.state`：`open` / `connecting` / `closing` / `closed`

#快速开始

阻塞式收发（适合脚本）：

python

import websocket

ws = websocket.connect("wss://echo.websocket.org")
ws.send("Hello")
msg = ws.receive(timeout=10)
print(msg)  # {"type": "text", "data": "Hello"}
ws.close()

回调式（适合 AppUI）：

python

import websocket


def on_message(data):
    print("收到:", data)


def on_close(code):
    print("关闭:", code)


ws = websocket.connect("wss://echo.websocket.org")
ws.on_message = on_message
ws.on_close = on_close
ws.send("Hello")

#AppUI 示例

连接、发送、断开放在按钮回调；最近一条消息展示在界面。

python

import appui
import websocket

ECHO_URL = "wss://echo.websocket.org"

state = appui.State(
    conn_state="未连接",
    last_msg="—",
    status="点击连接",
)

session = {"ws": None}


def on_message(data):
    state.last_msg = str(data)[:120]
    state.status = "收到消息"


def on_close(code):
    state.batch_update(conn_state="已关闭", status=f"关闭码 {code}")
    session["ws"] = None


def connect_echo():
    if session.get("ws") and session["ws"].is_open:
        state.status = "已连接"
        return

    try:
        ws = websocket.connect(ECHO_URL)
        ws.on_message = on_message
        ws.on_close = on_close
        session["ws"] = ws
        state.batch_update(conn_state=ws.state, status="已连接 echo 服务")
    except Exception as exc:
        state.batch_update(conn_state="失败", status=str(exc))


def send_ping():
    ws = session.get("ws")
    if not ws or not ws.is_open:
        state.status = "请先连接"
        return

    ws.send("ping from PythonIDE")
    state.status = "已发送 ping"


def disconnect():
    ws = session.get("ws")
    if ws:
        ws.close()
    session["ws"] = None
    state.conn_state = "未连接"


def body():
    return appui.NavigationStack(
        appui.Form([
            appui.Section("连接", [
                appui.Button("连接 Echo 服务", action=connect_echo)
                .button_style("bordered_prominent"),
                appui.Button("发送 ping", action=send_ping),
                appui.Button("断开", action=disconnect, role="destructive"),
            ]),
            appui.Section("状态", [
                appui.LabeledContent("连接", value=state.conn_state),
                appui.LabeledContent("最近消息", value=state.last_msg),
                appui.Text(state.status).foreground_color("secondaryLabel"),
            ]),
        ]).navigation_title("WebSocket")
    )


appui.run(body, state=state)

#API 参考

#速查

API	作用
`connect(url, protocols=None)`	建立连接，返回 `WebSocket`
`ws.send(data)`	发送文本或 `bytes`
`ws.receive(timeout=30)`	阻塞等待消息 → `dict` 或 `None`
`ws.close(code=1000)`	关闭连接
`ws.state` / `ws.is_open`	连接状态
`ws.on_message` / `on_open` / `on_close` / `on_error`	事件回调

#连接

python

ws = websocket.connect("wss://example.com/socket", protocols=["chat"])

连接失败抛 ConnectionError。

#收发

send(data) — 字符串发文本；bytes 发二进制（内部 Base64 编码）。

receive(timeout=30) — 返回 {"type": "text"|"binary", "data": ...} 或超时 None。

#回调

在 AppUI 中设置 on_message 等属性时，底层通过 _appui_native._register_callback 注册；需保持 ws 对象存活。

#常见错误

错误写法	后果	修正
在 `body()` 里 `connect`	每次刷新新建连接	放进按钮回调，复用 `session`
主线程长时间 `receive`	界面卡住	用 `on_message` 或后台线程
用 WebSocket 访问 REST	协议不匹配	用 network
忘记 `close()`	泄漏连接	用完关闭或依赖析构

#相关文档

文档	用途
network	HTTP 请求
http_server	本地 HTTP 服务（非 WS 服务端）

#预期效果

运行示例后，界面应出现文档描述的目标结果；若与预期不符，先看「失败路径」并按返回值或日志排查。