活动巴士:
# eventbus.py
EventKey = Union[int, str]
listeners: Dict[EventKey, Set[Callable]] = defaultdict(set)
def on(event: EventKey):
def decorator(callback):
listeners[event].add(callback)
@functools.wraps(callback)
def wrapper(*args, **kwargs):
return callback(*args, **kwargs)
return wrapper
return decorator
def emit(event: EventKey, *args, **kwargs):
for listener in listeners[event]:
listener(*args, **kwargs)
需要监听事件的类示例:
class Ticking(Referencable):
def __init__(self, id_: int):
super().__init__(id_)
@eventbus.on(StandardObjects.E_TIMER)
def on_clock(self, clock: Clock):
match clock.id:
case StandardObjects.TIME_TICK:
self.on_time_tick(clock)
def on_time_tick(self, clock: Clock):
pass
相关事件调用示例:
eventbus.emit(StandardObjects.E_TIMER, clock) # clock is an instance of Clock
我正在尝试在Python 3.11中编写一个相对简单的全局事件总线,但是,我想通过装饰器将侦听器注册到总线。下面的实现在装饰函数时工作正常,但在装饰类方法时会失败,因为调用时缺少“self”参数:
Ticking.on_clock() missing 1 required positional argument: 'clock'
(我可以确认这与“self”有关,因为将
listener(*args, **kwargs)
中的 emit()
修改为 listener('dummy', *args, **kwargs)
会引发预期的 AttributeError: 'str' object has no attribute 'on_time_tick'
。)
然后我探索了让装饰器以某种方式获取回调类实例的引用的方法,但在 Python 3 中,
Callable
对象不再有办法访问有关它们所属的类实例的元数据,而不是不稳定的特定于实现的反射 我当然想避免的黑客行为。
但是,如果您想要实例化对象的引用而不是类定义,则需要在创建类的实例时(也就是说,当您调用您时)修饰类的方法
__init__
方法。对代码进行以下修改应该可以解决您的问题
class Ticking(Referencable):
def __init__(self, id_: int):
super().__init__(id_)
# decorating the method self.on_clock after class intanciated
self.on_clock = eventbus.on(StandardObjects.E_TIMER)(self.on_clock)
def on_clock(self, clock: Clock):
match clock.id:
case StandardObjects.TIME_TICK:
self.on_time_tick(clock)
def on_time_tick(self, clock: Clock):
pass