你可以看看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"

如果你需要一个跨进程或网络边界的事件总线，你可以尝试PyMQ。它目前支持发布/订阅、消息队列和同步RPC。默认版本工作在一个Redis后端，所以你需要一个运行的Redis服务器。还有一个用于测试的内存后端。您还可以编写自己的后端。

import pymq

# common code
class MyEvent:
    pass

# subscribe code
@pymq.subscriber
def on_event(event: MyEvent):
    print('event received')

# publisher code
pymq.publish(MyEvent())

# you can also customize channels
pymq.subscribe(on_event, channel='my_channel')
pymq.publish(MyEvent(), channel='my_channel')

初始化系统。

from pymq.provider.redis import RedisConfig

# starts a new thread with a Redis event loop
pymq.init(RedisConfig())

# main application control loop

pymq.shutdown()

声明:我是这个库的作者

如果我在pyQt中做代码，我使用QT套接字/信号范式，django也是如此

如果我正在做异步I/O，我使用本机选择模块

如果我使用SAX python解析器，我使用SAX提供的事件API。所以看起来我是底层API的受害者:-)

也许你应该问问自己，你对事件框架/模块有什么期望。我个人倾向于使用qt中的Socket/Signal范例。更多关于这方面的信息可以在这里找到

PyPI packages

截至2022年10月，这些是PyPI上可用的与事件相关的软件包， 按最近发布日期订购。

PyDispatcher 2.0.6: Aug 2022 眨眼器1.5:2022年6月 pymitter 0.4.0: 2022年6月 python-dispatch 0.2.0: 2022年4月 插件1.0.0:2021年8月 事件0.4:2020年10月 zope。事件4.5.0:2020年9月 RxPy3 1.0.1: 2020年6月 路易2.0:2019年9月 PyPubSub 4.0.3: 2019年1月 Pyeventdispatcher 0.2.3a0: 2018 公交车道0.0.5:2018 PyPyDispatcher 2.1.2: 2017 Axel 0.0.7: 2016 Dispatcher 1.0: 2012 Py-notify 0.3.1: 2008

有更多的

有很多库可供选择，使用非常不同的术语(事件、信号、处理程序、方法分派、钩子等等)。

我试图对上述软件包以及这里的答案中提到的技术进行概述。

首先，一些术语……

观察者模式

事件系统最基本的风格是“处理程序方法包”，这是一个 观察者模式的简单实现。

基本上，处理程序方法(可调用对象)存储在一个数组中，每个方法在事件“触发”时被调用。

发布-订阅

观察者事件系统的缺点是只能在实际事件上注册处理程序 对象(或处理程序列表)。因此在注册时，事件需要已经存在。

这就是为什么存在第二种类型的事件系统 发布-订阅模式。 这里，处理程序不在事件对象(或处理程序列表)上注册，而是在中央调度程序上注册。 此外，通知器只与调度程序通信。该听什么，该发布什么 由'signal'决定，它只不过是一个名称(字符串)。

调停者模式

您可能也会感兴趣:中介模式。

钩子

钩子系统通常用于应用程序插件的上下文中。的 应用程序包含固定的集成点(钩子)，每个插件都可以 连接到该钩子并执行某些操作。

其他“事件”

注意:线程。事件不是一个“事件系统” 在上述意义上。这是一个线程同步系统，其中一个线程等待，直到另一个线程“信号”事件对象。

网络消息传递库也经常使用术语“事件”;有时它们在概念上是相似的;有时不是。 它们当然可以跨越线程、进程和计算机边界。如见。 pyzmq pymq, Twisted, Tornado, gevent, eventlet。

弱引用

在Python中，保留对方法或对象的引用可以确保它不会被删除 被垃圾收集器。这可能是可取的，但也可能导致内存泄漏: 链接的处理程序从来都不是 清理干净。

一些事件系统使用弱引用而不是常规引用来解决这个问题。

一些关于各种图书馆的词汇

观察者风格的事件系统:

zope.event shows the bare bones of how this works (see Lennart's answer). Note: this example does not even support handler arguments. LongPoke's 'callable list' implementation shows that such an event system can be implemented very minimalistically by subclassing list. Felk's variation EventHook also ensures the signatures of callees and callers. spassig's EventHook (Michael Foord's Event Pattern) is a straightforward implementation. Josip's Valued Lessons Event class is basically the same, but uses a set instead of a list to store the bag, and implements __call__ which are both reasonable additions. PyNotify is similar in concept and also provides additional concepts of variables and conditions ('variable changed event'). Homepage is not functional. axel is basically a bag-of-handlers with more features related to threading, error handling, ... python-dispatch requires the even source classes to derive from pydispatch.Dispatcher. buslane is class-based, supports single- or multiple handlers and facilitates extensive type hints. Pithikos' Observer/Event is a lightweight design.

发布-订阅库:

blinker has some nifty features such as automatic disconnection and filtering based on sender. PyPubSub is a stable package, and promises "advanced features that facilitate debugging and maintaining topics and messages". pymitter is a Python port of Node.js EventEmitter2 and offers namespaces, wildcards and TTL. PyDispatcher seems to emphasize flexibility with regards to many-to-many publication etc. Supports weak references. louie is a reworked PyDispatcher and should work "in a wide variety of contexts". pypydispatcher is based on (you guessed it...) PyDispatcher and also works in PyPy. django.dispatch is a rewritten PyDispatcher "with a more limited interface, but higher performance". pyeventdispatcher is based on PHP's Symfony framework's event-dispatcher. dispatcher was extracted from django.dispatch but is getting fairly old. Cristian Garcia's EventManger is a really short implementation.

其他:

Pluggy包含一个pytest插件使用的钩子系统。 RxPy3实现了Observable模式，并允许合并事件，重试等。 Qt的信号和插槽可以从PyQt中获得 或PySide2。当在同一个线程中使用时，它们作为回调工作， 或者作为两个不同线程之间的事件(使用事件循环)。信号和槽有它们的限制 仅适用于从QObject派生的类的对象。

我们使用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

另一个方便的包是事件。它将核心封装到事件订阅和事件触发，感觉像是语言的“自然”部分。它似乎类似于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或文档中找到。