如何在 Python 中使用装饰器创建事件监听器？答案

【问题标题】：How do I make an event listener with decorators in Python?如何在 Python 中使用装饰器创建事件监听器？
【发布时间】：2022-12-05 02:09:56
【问题描述】：

我想做一个事件监听器像这样：

@some.event
async def on_ready(some_info):
    print(some_info)

@some.event
async def on_error(err):
    print(err)

因此，当某些东西准备就绪时，或者如果像 WebSockets 一样收到消息，请将其用于 Discord，因为某些信息仅在 Bot 被识别或就绪时可用

我见过类似的东西：

def add_listener(func, name):
   # ...

def remove_listener(func, name):
   # ...

但我真的不知道如何使用它或创建一个

【问题讨论】：

你能澄清一下你到底需要什么帮助吗？难道只是写一个调用add_listener的装饰器？或者您是否需要知道如何让侦听器真正发挥作用（这是一个更大的问题，并且可能取决于您的问题空间的更多细节，例如您必须处理什么样的事件）？
如果你明白我的意思，我想做一个类似于“我见过”的事件监听器。我将它用于 Discord WebSocket/API 顺便说一句
请编辑您的问题（和标签）以包含这是针对 Discord 的事实。

标签： python discord decorator

【解决方案1】：

快速示例：

################################################################################
# the code for the "framework"
event1_listeners = []
event2_listeners = []

def listen_event1(func):
    event1_listeners.append(func)
    return func

def listen_event2(func):
    event2_listeners.append(func)
    return func

def process_event(event):
    if event["type"] == 1:
        for func in event1_listeners:
            func(event)
    elif event["type"] == 2:
        for func in event2_listeners:
            func(event)
    else:
        raise NotImplementedError(f"{event['type']=!r}")

################################################################################
# your code
@listen_event1
def handle_event1_v1(event):
    print(f"handle_event1_v1 : {event!r}")

@listen_event1
def handle_event1_v2(event):
    print(f"handle_event1_v2 : {event!r}")

@listen_event2
def handle_event2(event):
    print(f"handle_event2 : {event!r}")

################################################################################
# the events processed by the framework
process_event({"type": 1, "msg": "hello"})
process_event({"type": 2, "msg": "world"})

handle_event1_v1 : {'type': 1, 'msg': 'hello'}
handle_event1_v2 : {'type': 1, 'msg': 'hello'}
handle_event2 : {'type': 2, 'msg': 'world'}

本质上，装饰器会将函数存储在某个地方，当接收到事件时，框架会迭代为其注册的函数。
动态删除侦听器基本上只是从列表中删除 func 引用。
在这种情况下，装饰器只是“加糖”到不必自己做event1_listeners.append(func)。

【讨论】：

【解决方案2】：

这是我快速编写的一个简单的基于类的解决方案：

class Event:
    def __init__(self):
        # Initialise a list of listeners
        self.__listeners = []
    
    # Define a getter for the 'on' property which returns the decorator.
    @property
    def on(self):
        # A declorator to run addListener on the input function.
        def wrapper(func):
            self.addListener(func)
            return func
        return wrapper
    
    # Add and remove functions from the list of listeners.
    def addListener(self,func):
        if func in self.__listeners: return
        self.__listeners.append(func)
    
    def removeListener(self,func):
        if func not in self.__listeners: return
        self.__listeners.remove(func)
    
    # Trigger events.
    def trigger(self,args = []):
        # Run all the functions that are saved.
        for func in self.__listeners: func(*args)

它允许您创建一个Event，函数可以“订阅”到：

evn = Event()

# Some code...

evn.trigger(['arg x','arg y'])

这些函数都可以使用装饰器订阅事件：

@evn.on
def some_function(x,y): pass

或者使用 addListener 方法：

def some_function(x,y): pass
evn.addListener(some_function)

您还可以删除侦听器：

evn.removeListener(some_function)

要创建类似于您要求的内容，您可以执行以下操作：

# some.py

from event import Event

class SomeClass:
    def __init__(self):
        # Private event variable
        self.__event = Event()
        # Public event variable (decorator)
        self.event = self.__event.on

some = SomeClass()

然后像这样使用它：

# main.py

from some import some

@some.event
async def on_ready(some_info):
    print(some_info)

@some.event
async def on_error(err):
    print(err)

【讨论】：