我知道pydispatcher,但是Python周围一定有其他与事件相关的包。
哪些库是可用的?
我对作为大型框架一部分的事件管理器不感兴趣,我宁愿使用可以轻松扩展的小型骨架解决方案。
我知道pydispatcher,但是Python周围一定有其他与事件相关的包。
哪些库是可用的?
我对作为大型框架一部分的事件管理器不感兴趣,我宁愿使用可以轻松扩展的小型骨架解决方案。
当前回答
你可以看看pymitter (pypi)。它是一个小的单文件(~250 loc)方法 提供名称空间、通配符和TTL。
这里有一个基本的例子:
from pymitter import EventEmitter
ee = EventEmitter()
# decorator usage
@ee.on("myevent")
def handler1(arg):
print "handler1 called with", arg
# callback usage
def handler2(arg):
print "handler2 called with", arg
ee.on("myotherevent", handler2)
# emit
ee.emit("myevent", "foo")
# -> "handler1 called with foo"
ee.emit("myotherevent", "bar")
# -> "handler2 called with bar"
其他回答
另一个方便的包是事件。它将核心封装到事件订阅和事件触发,感觉像是语言的“自然”部分。它似乎类似于c#语言,后者提供了一种方便的方式来声明、订阅和触发事件。从技术上讲,事件是一个“槽”,可以将回调函数(事件处理程序)附加到其中—称为订阅事件的过程。
# Define a callback function
def something_changed(reason):
print "something changed because %s" % reason
# Use events module to create an event and register one or more callback functions
from events import Events
events = Events()
events.on_change += something_changed
触发事件时,将按顺序调用所有附加的事件处理程序。要触发该事件,请对槽位执行调用:
events.on_change('it had to happen')
这将输出:
'something changed because it had to happen'
更多的文档可以在github repo或文档中找到。
我使用zope.event。这是你能想象到的最简单的骨架。: -) 事实上,这里是完整的源代码:
subscribers = []
def notify(event):
for subscriber in subscribers:
subscriber(event)
例如,请注意不能在进程之间发送消息。它不是一个消息系统,只是一个事件系统,仅此而已。
如果我在pyQt中做代码,我使用QT套接字/信号范式,django也是如此
如果我正在做异步I/O,我使用本机选择模块
如果我使用SAX python解析器,我使用SAX提供的事件API。所以看起来我是底层API的受害者:-)
也许你应该问问自己,你对事件框架/模块有什么期望。我个人倾向于使用qt中的Socket/Signal范例。更多关于这方面的信息可以在这里找到
我们使用Michael ford在他的事件模式中建议的EventHook:
只需添加EventHooks到你的类:
class MyBroadcaster()
def __init__():
self.onChange = EventHook()
theBroadcaster = MyBroadcaster()
# add a listener to the event
theBroadcaster.onChange += myFunction
# remove listener from the event
theBroadcaster.onChange -= myFunction
# fire event
theBroadcaster.onChange.fire()
我们添加了从对象中移除所有侦听器到Michaels类的功能,最终得到了这样的结果:
class EventHook(object):
def __init__(self):
self.__handlers = []
def __iadd__(self, handler):
self.__handlers.append(handler)
return self
def __isub__(self, handler):
self.__handlers.remove(handler)
return self
def fire(self, *args, **keywargs):
for handler in self.__handlers:
handler(*args, **keywargs)
def clearObjectHandlers(self, inObject):
for theHandler in self.__handlers:
if theHandler.im_self == inObject:
self -= theHandler
这里是一个最小的设计,应该工作得很好。你要做的就是简单地在一个类中继承Observer,然后使用observe(event_name, callback_fn)来监听特定的事件。当特定事件在代码中的任何地方被触发时(例如。事件('USB connected')),则将触发相应的回调。
class Observer():
_observers = []
def __init__(self):
self._observers.append(self)
self._observed_events = []
def observe(self, event_name, callback_fn):
self._observed_events.append({'event_name' : event_name, 'callback_fn' : callback_fn})
class Event():
def __init__(self, event_name, *callback_args):
for observer in Observer._observers:
for observable in observer._observed_events:
if observable['event_name'] == event_name:
observable['callback_fn'](*callback_args)
例子:
class Room(Observer):
def __init__(self):
print("Room is ready.")
Observer.__init__(self) # DON'T FORGET THIS
def someone_arrived(self, who):
print(who + " has arrived!")
# Observe for specific event
room = Room()
room.observe('someone arrived', room.someone_arrived)
# Fire some events
Event('someone left', 'John')
Event('someone arrived', 'Lenard') # will output "Lenard has arrived!"
Event('someone Farted', 'Lenard')