如何在 python-socketio 服务器中使用自定义装饰器？

问题描述

我想在事件处理程序上使用一些自定义装饰器，以便更舒适地处理会话。

这是我没有自定义装饰器的事件处理程序：

@sio.event
async def test(sid,data):
    async with sio.session(sid) as sess:
        room_id = sess["room_id"]
        print('test',room_id)
        await sio.emit("test",{'msg': 'msg'},room=room_id)

这是我使用自定义装饰器的事件：

import functools


async def sess_decorator(func):
    @functools.wraps(func)
    async def wrapper(sid,data,*args,**kwargs):
        async with sio.session(sid) as sess:
            return func(sid,sess,**kwargs)
        logger.error(f'Failed to get sess.\nsid: {sid}\ndata: {data}')

    return wrapper


@sio.event
@sess_decorator
async def test(sid,sess):
     room_id = sess["room_id"]
     print('test',room_id)
     await sio.emit("test",room=room_id)

问题是在应用 @sess_decorator 事件后似乎根本没有触发。如果我更改装饰器的顺序，它会在调用我的装饰器之前失败并显示以下回溯：

Traceback (most recent call last):
  File "/home/pata/.pyenv/versions/3.7.7/envs/cartel_env/lib/python3.7/site-packages/socketio/asyncio_server.py",line 476,in _handle_event_internal
    r = await server._trigger_event(data[0],namespace,sid,*data[1:])
  File "/home/pata/.pyenv/versions/3.7.7/envs/cartel_env/lib/python3.7/site-packages/socketio/asyncio_server.py",line 504,in _trigger_event
    ret = await self.handlers[namespace][event](*args)
TypeError: test() missing 1 required positional argument: 'sess'

似乎有一些库设计问题？有没有其他方法可以完成相同的逻辑？

解决方法

您在实现装饰器方面有几个错误：

• 装饰器本身不应该是一个异步函数
• func 参数是原始函数，它是一个协程，所以应该等待

修复后的装饰器如下所示：

def sess_decorator(func):
    @functools.wraps(func)
    async def wrapper(sid,data,*args,**kwargs):
        async with sio.session(sid) as sess:
            return await func(sid,sess,**kwargs)
        logger.error(f'Failed to get sess.\nsid: {sid}\ndata: {data}')

    return wrapper