Python异步WebSocket出现AttributeError: __aenter__问题求助

问题原因 #

你当前使用的websocket.create_connection是同步WebSocket库的方法，返回的连接对象不支持异步上下文管理协议（未实现__aenter__和__aexit__异步方法），因此无法用async with语法包裹调用。非异步模式下直接调用同步方法可正常运行，但异步场景下必须使用适配异步IO的WebSocket库。

解决方案 #

1. 替换异步WebSocket库：使用官方异步库websockets（复数形式）替代原同步websocket库
2. 修正连接逻辑：用websockets.connect()替代create_connection()，该方法返回支持异步上下文管理的对象
3. 统一配置节点引用：代码中vehichle_counting_cam_1与Hydra配置里的vehichle_counting节点不匹配，需修正命名一致

修改后的核心代码（object_counting.py） #

from omegaconf import DictConfig
from pyrootutils import setup_root
from types import SimpleNamespace
import websockets  # 替换原同步websocket库
import dill as pickle
import asyncio
import hydra
import json

root = setup_root(
    search_from=__file__,
    indicator=[".git", "pyproject.toml"],
    pythonpath=True,
    dotenv=True,
)

# internal package
import websocket_data as wsd
from src.config.database_config import Database
from src.infra import time_infra as ABTime

class ObjectCounting:

    def __init__(self, config: DictConfig, ip_camera: str, ws_connection: str):
        self.config = config
        self.ip_camera = ip_camera
        self.ws_connection = ws_connection  # websocket


    def async_runner(self):
        asyncio.run(self.vehicle_async())
        
    async def vehicle_async(self):
        # database
        self.local_db = Database(
            self.config.local_db.host,
            self.config.local_db.port,
            self.config.local_db.user,
            self.config.local_db.password,
            self.config.local_db.database,
            self.config.local_db.db_type,
        ) if self.config.local_db.enabled else None
        
        # 修正配置节点引用，与Hydra配置统一
        self.push_table = self.config.service.vehichle_counting.table
        
        # 使用异步websockets库的上下文管理器
        async with websockets.connect(self.ws_connection) as async_ws_connection:
            while True:
                try:
                    # 异步接收WebSocket数据
                    raw_data = await async_ws_connection.recv()
                    json_data = json.loads(raw_data, object_hook=lambda d: SimpleNamespace(**d))
                    if json_data.type == "anpr":
                        raw_data = wsd.WebsocketData(json_data.type, 
                                                 json_data.payload, 
                                                 json_data.img)
                        payload_ws = raw_data.payload[0]
                        result_json = payload_ws.resjson
                        image_ws = raw_data.img
                        if result_json != '':
                            print("\ncatch!")
                            print('\nPayload: {}'.format(payload_ws))
                            response = {"camera_id": 123, 
                                        "camera_name": "test_camera", 
                                        "plate_number": payload_ws.label,
                                        "vehicle_type": payload_ws.vehicle_type,
                                        "datetime": ABTime.ab_timestamp(),
                                        "image": image_ws,
                                        }
                            self.local_db.push_data(self.push_table, response, self.local_db.db_type)

                except Exception as e:
                    print("error :", e)


if __name__ == "__main__":
    
    @hydra.main(config_path=root / "config", config_name="object_counter", version_base=None)
    def object_counter(hydra_config: DictConfig):
        # 修正配置节点引用
        hydra_service = hydra_config.service.vehichle_counting
        ip_camera = hydra_service.ip_cam
        ws_connection = hydra_service.ws_connect
        
        obj_counting = ObjectCounting(hydra_config, ip_camera, ws_connection)
        obj_counting.async_runner()
    object_counter()

额外注意事项 #

• 先卸载原同步库再安装异步库：pip uninstall websocket-client && pip install websockets
• 确保WebSocket服务端支持标准WebSocket协议，避免协议不兼容问题
• 若数据库操作是同步实现，建议封装为异步方法，避免阻塞异步事件循环

内容的提问来源于stack exchange，提问作者Moch. Chamdani M