Sistem acara apa untuk Python yang Anda gunakan? Saya sudah mengetahui pydispatcher , tetapi saya bertanya-tanya apa lagi yang bisa ditemukan, atau yang biasa digunakan?

Saya tidak tertarik pada manajer acara yang merupakan bagian dari kerangka kerja besar, saya lebih suka menggunakan solusi sederhana yang mudah saya perluas.

Paket PyPI

Pada Juni 2020, ini adalah paket terkait acara yang tersedia di PyPI, dipesan oleh tanggal rilis terbaru.

• RxPy3 1.0.1 : Juni 2020
• pluggy 0.13.1 : Juni 2020 (beta)
• Louie 2.0 : September 2019
• python-dispatch 0.1.2 : Feb 2019
• PyPubSub 4.0.3 : Jan 2019
• zope.event 4.4 : 2018
• pyeventdispatcher 0.2.3a0 : 2018
• buslane 0.0.5 : 2018
• PyPyDispatcher 2.1.2 : 2017
• axel 0.0.7 : 2016
• tanda bahayanya 1.4 : 2015
• PyDispatcher 2.0.5 : 2015
• pymitter 0.2.3 : 2014
• operator 1.0 : 2012
• py-notify 0.3.1 : 2008

Masih ada lagi

Itu banyak perpustakaan untuk dipilih, menggunakan terminologi yang sangat berbeda (peristiwa, sinyal, penangan, pengiriman metode, kait, ...).

Saya mencoba untuk menjaga ikhtisar paket di atas, ditambah teknik yang disebutkan dalam jawaban di sini.

Pertama, beberapa terminologi ...

Pola pengamat

Gaya paling dasar dari sistem acara adalah 'metode kantong penangan', yang merupakan implementasi sederhana dari pola Observer .

Pada dasarnya, metode handler (callable) disimpan dalam array dan masing-masing dipanggil ketika acara 'kebakaran'.

Publikasikan-Berlangganan

Kerugian dari sistem acara Pengamat adalah bahwa Anda hanya dapat mendaftarkan penangan pada objek Acara yang sebenarnya (atau daftar penangan). Jadi pada saat pendaftaran acara sudah perlu ada.

Karena itulah ada gaya kedua sistem acara: pola terbitkan-berlangganan . Di sini, penangan tidak mendaftar pada objek acara (atau daftar penangan), tetapi pada pengirim pusat. Pemberitahu juga hanya berbicara dengan operator. Apa yang harus didengarkan, atau apa yang akan diterbitkan ditentukan oleh 'sinyal', yang tidak lebih dari sebuah nama (string).

Pola mediator

Mungkin juga menarik: pola Mediator .

Kait

Sistem 'hook' biasanya digunakan dalam konteks plugin aplikasi. Aplikasi berisi titik integrasi tetap (kait), dan setiap plugin dapat terhubung ke kait itu dan melakukan tindakan tertentu.

'Acara' lainnya

Catatan: threading.Event bukan 'sistem acara' dalam arti di atas. Ini adalah sistem sinkronisasi utas di mana satu utas menunggu hingga utas lain 'memberi sinyal' pada objek Peristiwa.

Perpustakaan perpesanan jaringan sering menggunakan istilah 'acara' juga; terkadang konsepnya serupa; terkadang tidak. Mereka tentu saja dapat melintasi batas thread, proses, dan komputer. Lihat misalnya pyzmq , pymq , Twisted , Tornado , gevent , eventlet .

Referensi yang lemah

Dalam Python, memegang referensi ke metode atau objek memastikan bahwa itu tidak akan dihapus oleh pengumpul sampah. Ini dapat diinginkan, tetapi juga dapat menyebabkan kebocoran memori: penangan yang terhubung tidak pernah dibersihkan.

Beberapa sistem acara menggunakan referensi yang lemah alih-alih yang biasa untuk menyelesaikannya.

Beberapa kata tentang berbagai perpustakaan

Sistem acara bergaya pengamat:

• zope.event menunjukkan tulang kosong bagaimana ini bekerja (lihat jawaban Lennart ). Catatan: contoh ini bahkan tidak mendukung argumen penangan.
• Implementasi 'callable list' LongPoke menunjukkan bahwa sistem acara seperti itu dapat diimplementasikan dengan sangat minimalis dengan subklasifikasi list.
• Variasi Felk, EventHook juga memastikan tanda tangan callees dan penelepon.
• EventHook spassig (Pola Acara Michael Foord) adalah implementasi langsung.
• Kelas Event Pelajaran Nilai Josip pada dasarnya sama, tetapi menggunakan setbukan listuntuk menyimpan tas, dan mengimplementasikan __call__yang merupakan tambahan yang masuk akal.
• PyNotify serupa dalam konsep dan juga menyediakan konsep-konsep tambahan tentang variabel dan kondisi ('variabel berubah acara'). Beranda tidak berfungsi.
• axel pada dasarnya adalah kantung penanganan dengan lebih banyak fitur yang terkait dengan threading, penanganan kesalahan, ...
• python-dispatch membutuhkan kelas sumber genap untuk diturunkan pydispatch.Dispatcher.
• buslane berbasis kelas, mendukung penangan tunggal atau ganda dan memfasilitasi petunjuk jenis yang luas.
• Pengamat / Acara Pithikos adalah desain yang ringan.

Publikasikan-berlangganan perpustakaan:

• blinker memiliki beberapa fitur bagus seperti pemutusan sambungan otomatis dan penyaringan berdasarkan pengirim.
• PyPubSub adalah paket stabil, dan menjanjikan "fitur-fitur canggih yang memfasilitasi debugging dan menjaga topik dan pesan".
• pymitter adalah port Python dari Node.js EventEmitter2 dan menawarkan ruang nama, wildcard, dan TTL.
• PyDispatcher tampaknya menekankan fleksibilitas sehubungan dengan publikasi banyak ke banyak dll. Mendukung referensi yang lemah.
• Louie adalah PyDispatcher yang dikerjakan ulang dan harus bekerja "dalam berbagai konteks".
• pypydispatcher didasarkan pada (Anda dapat menebaknya ...) PyDispatcher dan juga berfungsi di PyPy.
• django.dispatch adalah PyDispatcher yang ditulis ulang "dengan antarmuka yang lebih terbatas, tetapi kinerjanya lebih tinggi".
• pyeventdispatcher didasarkan pada event-dispatcher framework PHP milik Symfony.
• dispatcher diekstraksi dari django.dispatch tetapi semakin tua.
• EventManger Cristian Garcia adalah implementasi yang sangat singkat.

Lainnya:

• Pluggy berisi sistem kait yang digunakan oleh pytestplugin.
• RxPy3 mengimplementasikan pola yang Dapat Diamati dan memungkinkan penggabungan acara, coba lagi dll.
• Sinyal dan Slot Qt tersedia dari PyQt atau PySide2 . Mereka berfungsi sebagai callback ketika digunakan di utas yang sama, atau sebagai peristiwa (menggunakan loop peristiwa) antara dua utas yang berbeda. Sinyal dan Slot memiliki batasan bahwa mereka hanya bekerja pada objek dari kelas yang berasal QObject.

Saya telah melakukannya dengan cara ini:

class Event(list):
    """Event subscription.

    A list of callable objects. Calling an instance of this will cause a
    call to each item in the list in ascending order by index.

    Example Usage:
    >>> def f(x):
    ...     print 'f(%s)' % x
    >>> def g(x):
    ...     print 'g(%s)' % x
    >>> e = Event()
    >>> e()
    >>> e.append(f)
    >>> e(123)
    f(123)
    >>> e.remove(f)
    >>> e()
    >>> e += (f, g)
    >>> e(10)
    f(10)
    g(10)
    >>> del e[0]
    >>> e(2)
    g(2)

    """
    def __call__(self, *args, **kwargs):
        for f in self:
            f(*args, **kwargs)

    def __repr__(self):
        return "Event(%s)" % list.__repr__(self)

Namun, seperti semua yang saya lihat, tidak ada pydoc yang dibuat secara otomatis untuk ini, dan tidak ada tanda tangan, yang benar-benar menyebalkan.

Kami menggunakan EventHook seperti yang disarankan dari Michael Foord dalam Pola Kegiatannya :

Cukup tambahkan EventHooks ke kelas Anda dengan:

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()

Kami menambahkan fungsionalitas untuk menghapus semua pendengar dari objek ke kelas Michaels dan berakhir dengan ini:

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

Saya menggunakan zope.event . Ini adalah tulang paling telanjang yang bisa Anda bayangkan. :-) Sebenarnya, ini adalah kode sumber lengkap:

subscribers = []

def notify(event):
    for subscriber in subscribers:
        subscriber(event)

Perhatikan bahwa Anda tidak dapat mengirim pesan antar proses, misalnya. Ini bukan sistem pesan, hanya sistem acara, tidak lebih, tidak kurang.

Saya menemukan skrip kecil ini di Valued Lessons . Tampaknya memiliki kesederhanaan / rasio daya yang tepat yang saya cari. Peter Thatcher adalah penulis kode berikut (tidak ada lisensi yang disebutkan).

class Event:
    def __init__(self):
        self.handlers = set()

    def handle(self, handler):
        self.handlers.add(handler)
        return self

    def unhandle(self, handler):
        try:
            self.handlers.remove(handler)
        except:
            raise ValueError("Handler is not handling this event, so cannot unhandle it.")
        return self

    def fire(self, *args, **kargs):
        for handler in self.handlers:
            handler(*args, **kargs)

    def getHandlerCount(self):
        return len(self.handlers)

    __iadd__ = handle
    __isub__ = unhandle
    __call__ = fire
    __len__  = getHandlerCount

class MockFileWatcher:
    def __init__(self):
        self.fileChanged = Event()

    def watchFiles(self):
        source_path = "foo"
        self.fileChanged(source_path)

def log_file_change(source_path):
    print "%r changed." % (source_path,)

def log_file_change2(source_path):
    print "%r changed!" % (source_path,)

watcher              = MockFileWatcher()
watcher.fileChanged += log_file_change2
watcher.fileChanged += log_file_change
watcher.fileChanged -= log_file_change2
watcher.watchFiles()

Berikut ini adalah desain minimal yang harus bekerja dengan baik. Yang harus Anda lakukan adalah mewarisiObserver kelas dan kemudian menggunakannya observe(event_name, callback_fn)untuk mendengarkan acara tertentu. Kapan saja peristiwa spesifik itu dipecat di mana saja dalam kode (mis. Event('USB connected')), Panggilan balik yang sesuai akan menyala.

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)

Contoh:

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')

Saya membuat EventManagerkelas (kode di akhir). Sintaksnya adalah sebagai berikut:

#Create an event with no listeners assigned to it
EventManager.addEvent( eventName = [] )

#Create an event with listeners assigned to it
EventManager.addEvent( eventName = [fun1, fun2,...] )

#Create any number event with listeners assigned to them
EventManager.addEvent( eventName1 = [e1fun1, e1fun2,...], eventName2 = [e2fun1, e2fun2,...], ... )

#Add or remove listener to an existing event
EventManager.eventName += extra_fun
EventManager.eventName -= removed_fun

#Delete an event
del EventManager.eventName

#Fire the event
EventManager.eventName()

Berikut ini sebuah contoh:

def hello(name):
    print "Hello {}".format(name)
    
def greetings(name):
    print "Greetings {}".format(name)

EventManager.addEvent( salute = [greetings] )
EventManager.salute += hello

print "\nInitial salute"
EventManager.salute('Oscar')

print "\nNow remove greetings"
EventManager.salute -= greetings
EventManager.salute('Oscar')

Keluaran:

Salam hormat, Salam.
Halo Oscar

Sekarang hapus salam
Halo Oscar

Kode EventManger:

class EventManager:
    
    class Event:
        def __init__(self,functions):
            if type(functions) is not list:
                raise ValueError("functions parameter has to be a list")
            self.functions = functions
            
        def __iadd__(self,func):
            self.functions.append(func)
            return self
            
        def __isub__(self,func):
            self.functions.remove(func)
            return self
            
        def __call__(self,*args,**kvargs):
            for func in self.functions : func(*args,**kvargs)
            
    @classmethod
    def addEvent(cls,**kvargs):
        """
        addEvent( event1 = [f1,f2,...], event2 = [g1,g2,...], ... )
        creates events using **kvargs to create any number of events. Each event recieves a list of functions,
        where every function in the list recieves the same parameters.
        
        Example:
        
        def hello(): print "Hello ",
        def world(): print "World"
        
        EventManager.addEvent( salute = [hello] )
        EventManager.salute += world
        
        EventManager.salute()
        
        Output:
        Hello World
        """
        for key in kvargs.keys():
            if type(kvargs[key]) is not list:
                raise ValueError("value has to be a list")
            else:
                kvargs[key] = cls.Event(kvargs[key])
        
        cls.__dict__.update(kvargs)

Anda mungkin melihat pymitter ( pypi ). Ini adalah satu file kecil (~ 250 loc) pendekatan "menyediakan ruang nama, wildcard dan TTL".

Inilah contoh dasar:

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"

Saya membuat variasi dari pendekatan minimalis Longpoke yang juga memastikan tanda tangan untuk callees dan penelepon:

class EventHook(object):
    '''
    A simple implementation of the Observer-Pattern.
    The user can specify an event signature upon inizializazion,
    defined by kwargs in the form of argumentname=class (e.g. id=int).
    The arguments' types are not checked in this implementation though.
    Callables with a fitting signature can be added with += or removed with -=.
    All listeners can be notified by calling the EventHook class with fitting
    arguments.

    >>> event = EventHook(id=int, data=dict)
    >>> event += lambda id, data: print("%d %s" % (id, data))
    >>> event(id=5, data={"foo": "bar"})
    5 {'foo': 'bar'}

    >>> event = EventHook(id=int)
    >>> event += lambda wrong_name: None
    Traceback (most recent call last):
        ...
    ValueError: Listener must have these arguments: (id=int)

    >>> event = EventHook(id=int)
    >>> event += lambda id: None
    >>> event(wrong_name=0)
    Traceback (most recent call last):
        ...
    ValueError: This EventHook must be called with these arguments: (id=int)
    '''
    def __init__(self, **signature):
        self._signature = signature
        self._argnames = set(signature.keys())
        self._handlers = []

    def _kwargs_str(self):
        return ", ".join(k+"="+v.__name__ for k, v in self._signature.items())

    def __iadd__(self, handler):
        params = inspect.signature(handler).parameters
        valid = True
        argnames = set(n for n in params.keys())
        if argnames != self._argnames:
            valid = False
        for p in params.values():
            if p.kind == p.VAR_KEYWORD:
                valid = True
                break
            if p.kind not in (p.POSITIONAL_OR_KEYWORD, p.KEYWORD_ONLY):
                valid = False
                break
        if not valid:
            raise ValueError("Listener must have these arguments: (%s)"
                             % self._kwargs_str())
        self._handlers.append(handler)
        return self

    def __isub__(self, handler):
        self._handlers.remove(handler)
        return self

    def __call__(self, *args, **kwargs):
        if args or set(kwargs.keys()) != self._argnames:
            raise ValueError("This EventHook must be called with these " +
                             "keyword arguments: (%s)" % self._kwargs_str())
        for handler in self._handlers[:]:
            handler(**kwargs)

    def __repr__(self):
        return "EventHook(%s)" % self._kwargs_str()

Jika saya melakukan kode di pyQt saya menggunakan paradigma QT soket / sinyal, hal yang sama juga berlaku untuk Django

Jika saya melakukan I / OI async gunakan modul pilih asli

Jika saya menggunakan parser SAX python, saya menggunakan API acara yang disediakan oleh SAX. Jadi sepertinya saya korban API yang mendasarinya :-)

Mungkin Anda harus bertanya pada diri sendiri apa yang Anda harapkan dari kerangka / modul acara. Preferensi pribadi saya adalah menggunakan paradigma Socket / Sinyal dari QT. info lebih lanjut tentang itu dapat ditemukan di sini

Berikut modul lain untuk dipertimbangkan. Tampaknya pilihan yang layak untuk aplikasi yang lebih banyak menuntut.

Py-notify adalah paket Python yang menyediakan alat untuk menerapkan pola pemrograman Observer. Alat-alat ini termasuk sinyal, kondisi dan variabel.

Sinyal adalah daftar penangan yang dipanggil saat sinyal dipancarkan. Kondisi pada dasarnya adalah variabel boolean ditambah dengan sinyal yang dipancarkan ketika kondisi kondisi berubah. Mereka dapat digabungkan menggunakan operator logika standar (bukan, dan, dll.) Ke dalam kondisi gabungan. Variabel, tidak seperti kondisi, dapat menampung objek Python apa pun, tidak hanya boolean, tetapi mereka tidak dapat digabungkan.

Jika Anda ingin melakukan hal-hal yang lebih rumit seperti menggabungkan acara atau coba lagi, Anda dapat menggunakan pola Observable dan perpustakaan dewasa yang mengimplementasikannya. https://github.com/ReactiveX/RxPY . Dapat diamati sangat umum dalam Javascript dan Java dan sangat nyaman digunakan untuk beberapa tugas async.

from rx import Observable, Observer


def push_five_strings(observer):
        observer.on_next("Alpha")
        observer.on_next("Beta")
        observer.on_next("Gamma")
        observer.on_next("Delta")
        observer.on_next("Epsilon")
        observer.on_completed()


class PrintObserver(Observer):

    def on_next(self, value):
        print("Received {0}".format(value))

    def on_completed(self):
        print("Done!")

    def on_error(self, error):
        print("Error Occurred: {0}".format(error))

source = Observable.create(push_five_strings)

source.subscribe(PrintObserver())

OUTPUT :

Received Alpha
Received Beta
Received Gamma
Received Delta
Received Epsilon
Done!

Jika Anda memerlukan eventbus yang berfungsi melintasi batas proses atau jaringan, Anda dapat mencoba PyMQ . Saat ini mendukung pub / sub, antrian pesan dan RPC sinkron. Versi default berfungsi di atas backend Redis, jadi Anda memerlukan server Redis yang sedang berjalan. Ada juga backend dalam memori untuk pengujian. Anda juga dapat menulis backend Anda sendiri.

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')

Untuk menginisialisasi sistem:

from pymq.provider.redis import RedisConfig

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

# main application control loop

pymq.shutdown()

Penafian: Saya penulis perpustakaan ini

Anda dapat mencoba buslanemodul.

Perpustakaan ini membuat implementasi sistem berbasis pesan lebih mudah. Ini mendukung perintah (penangan tunggal) dan pendekatan peristiwa (0 atau beberapa penangan). Buslane menggunakan anotasi jenis Python untuk mendaftarkan handler dengan benar.

Contoh sederhana:

from dataclasses import dataclass

from buslane.commands import Command, CommandHandler, CommandBus


@dataclass(frozen=True)
class RegisterUserCommand(Command):
    email: str
    password: str


class RegisterUserCommandHandler(CommandHandler[RegisterUserCommand]):

    def handle(self, command: RegisterUserCommand) -> None:
        assert command == RegisterUserCommand(
            email='[email protected]',
            password='secret',
        )


command_bus = CommandBus()
command_bus.register(handler=RegisterUserCommandHandler())
command_bus.execute(command=RegisterUserCommand(
    email='[email protected]',
    password='secret',
))

Untuk menginstal buslane, cukup gunakan pip:

$ pip install buslane

Beberapa waktu lalu saya sudah menulis perpustakaan yang mungkin berguna untuk Anda. Ini memungkinkan Anda untuk memiliki pendengar lokal dan global, berbagai cara untuk mendaftarkan mereka, prioritas eksekusi dan sebagainya.

from pyeventdispatcher import register

register("foo.bar", lambda event: print("second"))
register("foo.bar", lambda event: print("first "), -100)

dispatch(Event("foo.bar", {"id": 1}))
# first second

Silahkan lihat pyeventdispatcher