Python Websocket recv方法asyncio超时设置报错求助

错误原因 #

你遇到的TypeError: An asyncio.Future, a coroutine or an awaitable is required错误，核心原因是：

• 你使用的是websocket库的同步create_connection方法，返回的是同步WebSocket对象，它的recv()是阻塞式的同步方法，并非asyncio要求的协程/可等待对象
• asyncio.wait_for只能接收Future、协程或其他可等待对象，直接传入同步方法调用自然会报错

当前方案的问题 #

你的方案存在几个关键问题：

1. 同步与异步代码混合：同步WebSocket的阻塞IO操作会卡住整个asyncio事件循环，完全失去异步编程的优势
2. 同步上下文管理器不适配异步场景：自定义的__enter__/__exit__是同步方法，在异步函数中使用会导致连接建立、登录这些操作阻塞事件循环
3. 时间计算不可靠：用time.time()计算剩余超时时间，依赖系统时间，若系统时间被修改会导致超时逻辑出错

优化方案 #

改用官方异步WebSocket库websockets（注意带末尾的s），实现异步上下文管理器，全程使用异步非阻塞IO，让超时机制正常工作。

步骤1：安装依赖 #

pip install websockets

步骤2：修改后的代码 #

import websockets
import ssl
import json
import asyncio

class WebsocketSession:
    def __init__(self, username, password, environment, id):
        self.username = username
        self.password = password
        self.env = environment
        self.id = id
        self.ws = None
    
    def build_login_credentials(self):
        return {
            # 替换为你的登录JSON结构
        }
    
    def url(self):
        # 替换为你的实际WebSocket URL生成逻辑
        return f"wss://{self.env}.your-domain.com/ws"
    
    async def send(self, msg):
        await self.ws.send(json.dumps(msg))

    async def find(self, selection_func, timeout=10):
        # 使用asyncio单调时间，避免系统时间变动影响超时计算
        start_time = asyncio.get_event_loop().time()
        while True:
            elapsed = asyncio.get_event_loop().time() - start_time
            time_remaining = timeout - elapsed
            
            if time_remaining <= 0:
                raise asyncio.TimeoutError("未在超时时间内找到目标响应")
            
            # 包装异步recv方法实现超时
            resp_str = await asyncio.wait_for(self.ws.recv(), timeout=time_remaining)
            resp = json.loads(resp_str)

            if selection_func(resp):
                print(f"Received Response: {resp}")
                return resp
            else:
                print(f"Received Type: {resp['type']}")
    
    async def __aenter__(self):
        # 创建异步SSL上下文
        ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ssl_context.check_hostname = False
        ssl_context.verify_mode = ssl.CERT_NONE
        
        # 建立异步WebSocket连接
        self.ws = await websockets.connect(
            self.url(),
            ssl=ssl_context
        )
        # 发送登录凭证
        await self.send(self.build_login_credentials())

        # 接收并校验登录响应
        login_resp = json.loads(await self.ws.recv())
        # 替换为你的登录错误校验逻辑
        if login_resp.get("type") != "loginSuccess":
            raise Exception(f"登录失败: {login_resp}")

        return self
    
    async def __aexit__(self, *args):
        if self.ws:
            await self.ws.close()

async def main():
    username = "redacted"
    password = "redacted"
    env = "int"
    id = 123456

    # 使用async with配合异步上下文管理器
    async with WebsocketSession(username, password, env, id) as ws_session:
        config = await ws_session.find(lambda resp: resp["type"] == "configList")
        # 替换为你的业务逻辑
        print(f"获取到配置: {config}")

if __name__ == "__main__":
    asyncio.run(main())

优化点说明 #

• 全异步IO：websockets库的所有操作都是异步非阻塞的，不会卡住事件循环
• 异步上下文管理器：实现__aenter__/__aexit__方法，配合async with使用，符合异步代码规范
• 可靠的超时计算：用asyncio.get_event_loop().time()获取单调递增时间，不受系统时间修改影响
• 简洁的超时逻辑：直接通过asyncio.wait_for包装异步recv，无需手动计算剩余时间的循环（保留循环是为了持续过滤响应，逻辑合理）

内容的提问来源于stack exchange，提问作者T. Smith