求Python Requests库的Websockets适配器现成实现方案

现成实现 #

• 第三方库requests-ws提供了开箱即用的WebSocket适配器，完全兼容Requests的会话机制，支持ws://和wss://协议，可直接挂载到Requests会话中使用。
• 社区也有基于BaseAdapter的自定义实现，核心是用websockets库替换HTTP传输逻辑，适配Requests的接口规范。

自定义适配器示例 #

如果需要更灵活的控制，可自行实现适配器，以下是基础版本：

import requests
from requests.adapters import BaseAdapter
from requests.models import Response
import websockets
import asyncio

class WebSocketAdapter(BaseAdapter):
    def send(self, request, stream=False, timeout=None, verify=True, cert=None, proxies=None):
        async def handle_ws():
            async with websockets.connect(request.url, ssl=verify) as ws:
                # 发送请求体（如果存在）
                if request.body:
                    await ws.send(request.body)
                # 接收响应数据
                resp_data = await ws.recv()
                
                # 构造符合Requests规范的响应对象
                response = Response()
                response.status_code = 101  # WebSocket协议切换成功状态码
                response._content = resp_data.encode('utf-8')
                response.url = request.url
                response.headers['Content-Type'] = 'text/plain'
                return response
        
        # 同步执行异步WebSocket逻辑
        try:
            loop = asyncio.get_event_loop()
            if loop.is_running():
                response = loop.run_until_complete(asyncio.create_task(handle_ws()))
            else:
                response = loop.run_until_complete(handle_ws())
            return response
        except Exception as e:
            raise requests.exceptions.ConnectionError(str(e)) from e

# 会话挂载适配器
session = requests.Session()
session.mount('ws://', WebSocketAdapter())
session.mount('wss://', WebSocketAdapter())

适配Zeep的使用方式 #

Zeep底层依赖Requests传输，只需将挂载了WebSocket适配器的会话传给Zeep的Transport即可：

from zeep import Client
from zeep.transports import Transport

# 复用上面的session
transport = Transport(session=session)
client = Client('你的WSDL地址', transport=transport)

# 调用WebSocket端点的方法
result = client.service.target_method()

注意事项 #

• WebSocket是双向长连接协议，上述适配器仅处理单次请求-响应场景，若需持续通信，需扩展适配器逻辑（比如保持连接、处理消息流）。
• 异步逻辑兼容性：如果应用本身是异步架构，建议调整适配器实现方式，避免同步阻塞。
• 认证、超时等功能需自行扩展，比如在websockets.connect中添加认证参数，或通过asyncio.wait_for处理超时。

内容的提问来源于stack exchange，提问作者Cristi