Le système d'événements en Python

Enveloppant les différents systèmes à événements qui sont mentionnés dans les réponses ici:

Le plus de style de base de système d'événements est le "sac de méthodes de gestionnaire", qui est une simple mise en œuvre de la Modèle observateur. Fondamentalement, les méthodes de gestionnaire d' (callables) sont stockées dans un tableau et chacune est appelée lorsque l'événement "feux".

• zope.l'événement montre les os nus de la façon dont cela fonctionne (voir Lennart réponse). Remarque: cet exemple n'a même pas le soutien gestionnaire d'arguments.
• LongPoke du "appelable liste" mise en œuvre montre qu'un tel événement système peut être mis en œuvre très minimalistically par sous-classement list.
• spassig de EventHook (Michael Foord de l'Événement Pattern) est une mise en œuvre directe.
• Josip est Évalué les Leçons des Événements de la classe est fondamentalement la même, mais utilise un set au lieu d'un list pour stocker le sac, et met en œuvre __call__ qui sont à la fois raisonnable ajouts.
• PyNotify concept est similaire et fournit également d'autres notions de variables et de modalités (variable changé événement").
• axel est essentiellement un sac de gestionnaires avec plus de fonctionnalités liées à l'enfilage, erreur de manipulation, ...

L'inconvénient de ces systèmes est que vous ne pouvez enregistrer les gestionnaires sur l'Événement réel de l'objet (ou les gestionnaires de la liste).
Ainsi, au moment de l'enregistrement de l'événement a déjà besoin d'exister.

C'est pourquoi le deuxième style de systèmes à événements: la publish-subscribe modèle.
Ici, les gestionnaires ne vous inscrivez pas sur un objet événement (ou gestionnaire de liste), mais sur un répartiteur central. Aussi les auteurs de la notification de parler à l'expéditeur. Ce qu'il faut écouter, ou de quoi publier est déterminée par "signal", qui n'est rien de plus qu'un nom (chaîne de caractères).

• clignotant a quelques fonctionnalités intéressantes telles que la déconnexion automatique et filtrage basé sur l'expéditeur.
• PyPubSub , à première vue, semble être assez simple.
• PyDispatcher semble souligner la flexibilité en ce qui concerne plusieurs-à-plusieurs publications etc.
• louie est une version retravaillée de la PyDispatcher "fournir plugin infrastructure, y compris Tordu et PyQt de soutien spécifique". Il semble avoir perdu de maintenance à compter de janvier 2016.
• django.envoi est une réécrit PyDispatcher "avec un nombre plus limité d'interface, mais une augmentation de la performance".
• Qt sur les Signaux et les Slots sont disponibles à partir de PyQt ou PySide. Ils travaillent à titre de rappel lorsque utilisé dans le même thread, ou comme des événements (à l'aide d'une boucle d'événement) entre deux threads différents. Les signaux et les Slots sont à la limitation qu'ils travaillent seulement dans les objets de classes qui dérivent de QObject.

Remarque: le filetage.L'événement n'est pas un "système", dans le sens indiqué ci-dessus. C'est un thread système de synchronisation où un thread attend jusqu'à ce qu'un autre thread "signaux" de l'objet d'Événement.

Note: ne figurent pas ci-dessus sont pypydispatcher, python-expédition et le "crochet" de pluggy pourrait être d'intérêt.

Nous utilisons un EventHook comme suggéré de Michael Foord dans son Cas Motif:

Juste ajouter EventHooks à vos classes avec:

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

Nous ajouter la fonctionnalité de suppression de tous les écouteurs d'un objet à Michaels de classe et qui a abouti à ceci:

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

J'ai trouvé ce petit script sur Apprécié Des Leçons. Il semble avoir juste le droit de la simplicité/rapport de puissance je suis après. Peter Thatcher est l'auteur de code suivant (pas de licence est mentionné).

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

J'ai créé un EventManager classe (code à la fin). La syntaxe est la suivante:

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

Voici un Exemple:

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

De sortie:

Initiale salut

Salutations Oscar

Bonjour Oscar

Maintenant supprimer salutations

Bonjour Oscar

EventManger Code:

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)

J'ai fait une variation de Longpoke est minimaliste approche qui assure également la signature pour les deux fonctions appelées et les appelants:

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

Voici un autre module pour examen. Il semble un choix viable pour les applications les plus exigeantes.

Py-notifier est un paquet Python
fournir des outils pour la mise en œuvre de
Observateur de programmation de modèle. Ces
les outils incluent les signaux, les conditions et les
les variables.

Signaux sont des listes de gestionnaires qui sont
appelé lorsque le signal est émis.
Les Conditions sont essentiellement de type boolean
les variables couplé avec un signal
est émis lorsque la condition de l'état
les changements. Ils peuvent être combinés à l'aide de
norme d'opérateurs logiques (et,
etc.) composés conditions.
Les Variables, à la différence des conditions, peuvent tenir
tout objet Python, et pas seulement les booléens, les
mais ils ne peuvent pas être combinées.

Vous pouvez essayer buslane module.

Cette bibliothèque permet la mise en œuvre de message basé sur un système plus facile. Il prend en charge les commandes (seul gestionnaire) et les événements (0 ou plusieurs gestionnaires) approche. Buslane utilise Python annotations de type à enregistrer correctement gestionnaire.

Exemple Simple:

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

Pour installer buslane, il suffit d'utiliser pip:

$ pip install buslane