如何通过指定字符串为python-socketio动态添加事件处理器？

我有Python-SocketIO的服务端与客户端，逻辑是服务端以事件名（如ping_client）调用客户端，客户端收到后回发ping_client_response事件。

可正常运行的服务端代码 #

sio = socketio.AsyncServer(async_mode='asgi')

async def sio_request(event):
    @sio.event
    def ping_response(sid, file_pack):
        print('got response from client')

    await sio.emit(event, [])  

sio_request('ping_client')

这段代码里print('got response from client')能正常执行。

尝试动态添加处理器的失效代码 #

我想动态添加异步事件处理器，写了以下代码，但预期的handler_callback(sid, data)并未被调用：

sio = socketio.AsyncServer(async_mode='asgi')

async def sio_request(event, handler_callback):
    # 定义事件处理器生成器
    def create_event_handler(event_name):
        async def event_handler(sid, data):
            # 动态事件处理逻辑
            handler_callback(sid, data)
        return event_handler

    # 创建动态事件处理器
    dynamic_resp_handler = create_event_handler(event)

    func_name = event + '_response'
    # func_name = 'ping_client_response'
    setattr(sio, func_name, dynamic_resp_handler)

    await sio.emit(event, [])  

sio_request('ping_client')

客户端代码 #

@sio.event
def ping_client(data):
    print('ping_client from server')
    sio.emit('ping_client_response', 'ping_client_response from client')    

请问如何通过指定字符串为Python-SocketIO动态添加事件处理器？

问题原因 #

直接用setattr给sio对象绑定函数的方式，并没有让Python-SocketIO的事件注册机制识别到这个处理器。@sio.event装饰器内部不仅是把函数绑定到对象属性，还会将函数注册到服务端的事件路由表中。

解决方案 #

使用Python-SocketIO提供的sio.on()方法来动态注册事件处理器，而不是手动设置对象属性。

修改后的服务端代码：

import asyncio
import socketio

sio = socketio.AsyncServer(async_mode='asgi')

async def sio_request(event, handler_callback):
    async def event_handler(sid, data):
        # 兼容同步/异步回调
        if asyncio.iscoroutinefunction(handler_callback):
            await handler_callback(sid, data)
        else:
            handler_callback(sid, data)

    # 动态注册对应事件的处理器
    sio.on(event + '_response', event_handler)

    await sio.emit(event, [])  

# 示例调用：传入自定义回调函数
async def my_callback(sid, data):
    print(f"收到客户端响应：sid={sid}, data={data}")

await sio_request('ping_client', my_callback)

关键说明 #

1. 官方注册方式：sio.on()是Python-SocketIO官方支持的动态事件注册方法，会将处理器加入服务端的事件映射表，确保客户端发来的事件能被正确匹配。
2. 回调兼容性：加入异步判断逻辑，避免异步回调因未被await而无法执行完成。
3. 事件名一致性：确保注册的事件名与客户端回发的事件名完全匹配（这里为event + '_response'）。

如果后续需要移除该动态处理器，可使用sio.off()方法，传入对应的事件名和处理器函数。

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