Система событий в Python

Какую систему событий для Python вы используете? Я уже в курсе Pydispatcher, но мне было интересно, что еще можно найти или обычно используется?

Меня не интересуют менеджеры событий, которые являются частью больших фреймворков, я бы предпочел использовать небольшое решение, которое я могу легко расширить.

Ответы на вопрос(13)

Я использую Zope.event. Это самые голые кости, которые вы можете себе представить. :-) На самом деле, вот полный исходный код:

subscribers = []

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

Обратите внимание, что вы не можете отправлять сообщения между процессами, например. Это не система обмена сообщениями, просто система событий, ни больше, ни меньше.

 Lennart Regebro08 июл. 2009 г., 11:48
Вы можете отправить все, что захотите, с помощью zope.event. Но я хочу сказать, что это неправильная система обмена сообщениями, поскольку вы не можете отправлять события / сообщения другим процессам или другим компьютерам. Вероятно, вы должны быть более конкретными с вашими требованиями.
 Boldewyn07 июл. 2009 г., 17:03
Pypi.python.org / PyPI / zope.event ... чтобы спасти беднягу Гугл; -)
 Josip08 июл. 2009 г., 09:24
Я все еще хотел бы иметь возможность отправлять сообщения. Я бы использовал систему событий в приложении, построенном на Tkinter. Я не использую его систему событий, потому что она не поддерживает сообщения.
Решение Вопроса

которые упомянуты в ответах здесь:

Самым основным стилем системы событий является «сумка методов обработки», которая представляет собой простую реализацию Наблюдательный образец. По сути, методы-обработчики (callables) хранятся в массиве и каждый из них вызывается, когда событие «срабатывает».

Zope.event показывает, как это работает (см. Ответ Леннарта). Примечание: этот пример даже не поддерживает аргументы обработчика. Список вызываемых пользователей LongPoke реализация показывает, что такая система событий может быть реализована очень минималистично с помощью подклассовlist. Spassig's EventHook (Шаблон событий Майкла Фурда) - простая реализация. Ценные уроки Иоси в основном то же самое, но используетset вместоlist чтобы хранить сумку и инструменты__call__ которые являются разумными дополнениями. PyNotify похожа по своей концепции и также предоставляет дополнительные понятия о переменных и условиях («событие изменено переменной»). Аксель - это, по сути, мешок с обработчиками, с большим количеством функций, связанных с многопоточностью, обработкой ошибок, ...

Недостаток этих систем событий заключается в том, что вы можете зарегистрировать обработчики только для реального объекта события (или списка обработчиков). Таким образом, во время регистрации событие уже должно существовать.

Вот почему существует второй стиль систем событий: шаблон публикации-подписки. Здесь обработчики регистрируются не в объекте события (или списке обработчиков), а в центральном диспетчере. Также уведомители общаются только с диспетчером. Что слушать или что публиковать, определяется «сигналом», который является не чем иным, как именем (строкой).

Мигалка @ есть отличные функции, такие как автоматическое отключение и фильтрация по отправителю. PyPubSub с первого взгляда кажется довольно простым. PyDispatcher, кажется, подчеркивает гибкость в отношении публикации «многие ко многим» и т. д.Лу - это переработанный PyDispatcher, «обеспечивающий инфраструктуру плагинов, включая поддержку Twisted и PyQt». Похоже, что потерял обслуживание после января 2016 года. Django.dispatch - это переписанный PyDispatcher «с более ограниченным интерфейсом, но более высокой производительностью». Сигналы и слоты Qt доступны на PyQt или PySide. Они работают как обратный вызов, когда используются в одном потоке, или как события (используя цикл событий) между двумя различными потоками. Сигналы и слоты имеют ограничение, заключающееся в том, что они работают только в объектах классов, производных отQObject.

Заметка: Threading.Event не является «системой событий» в вышеприведенном смысле. Это система синхронизации потоков, в которой один поток ожидает, пока другой поток не «сигнализирует» объект Event.

Примечание: еще не включены выше Pypydispatcher, Питон-диспетчерская и «система крючков» из Pluggy тоже может быть интересным.

 Daniel Cardenas12 июл. 2017 г., 05:29
Это очень хорошо. большое спасибо
 florisla14 мар. 2016 г., 10:47
@ the979kid louie, похоже, плохо поддерживается, страница pypi ссылается на 404-е на GitHub: 11craft.github.io / Louie ; Github.com / gldnspud / Louie. Должно быть Github.com / 11craft / Louie.
 Oliver13 авг. 2013 г., 00:19
Вы правы насчет PyPubSub: пока не поддерживает Python3. Это в моем списке дел Sourceforge.net / р / PubSub / задачи).
 the979kid30 авг. 2015 г., 11:21
Есть также Луи, который основан на PyDispatcher: Pypi.python.org / PyPI / Louie / 1,1
 kxr11 апр. 2017 г., 13:44
лабые слушатели событий являются общей потребностью. В противном случае использование в реальном мире становится трудным. Обратите внимание, какие решения могут быть полезны.

Я делал это так:

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)

Однако, как и со всем, что я видел, для этого не существует автоматически сгенерированного pydoc и подписей, которые действительно отстой.

 akaRem08 февр. 2013 г., 15:35
Очень красивый минималистичный стиль! супер
 Gabe Spradlin04 дек. 2017 г., 17:48
@ omgimdrunk Простой обработчик событий будет запускать одну или несколько вызываемых функций при каждом событии. Класс, «управляющий» этим для вас, потребует как минимум следующих методов - add & fire. В этом классе вам нужно будет поддерживать список обработчиков, которые будут выполнены. Давайте поместим это в переменную экземпляра_bag_of_handlers который является списком. Метод add класса будет простоself._bag_of_handlers.append(some_callable). Метод огня класса будет проходить через _bag_of_handlers`, передавая предоставленные аргументы и kwargs обработчикам и выполняя каждый в последовательности.
 omgimdrunk28 сент. 2017 г., 18:15
большое одолжение, может кто-нибудь объяснить это, как мне было 10 лет? Этот класс наследуется основным классом? Я не вижув это так что super () не будет использоваться. По какой-то причине мне это не нужно.
 Rudy Lattae31 дек. 2010 г., 09:25
Я нахожу этот стиль довольно интригующим. Это сладко голые. Мне нравится тот факт, что он позволяет манипулировать событиями и их подписчиками как автономными операциями. Я посмотрю, как это будет в реальном проекте.
 user89016703 апр. 2014 г., 21:20
Я не могу сказать достаточно, это действительно просто и легко.

я использую парадигму сокетов / сигналов QT, то же самое относится и к django

Если я делаю асинхронный ввод-вывод, я использую собственный модуль выбора

Если я использую синтаксический анализатор Python SAX, я использую API событий, предоставляемый SAX. Похоже, я жертва API: -)

Может быть, вы должны спросить себя, что вы ожидаете от фреймворка / модуля событий. Лично я предпочитаю использовать парадигму Socket / Signal от QT. больше информации об этом можно найтиВо

который также обеспечивает подписи как для вызывающих, так и для вызывающих абонентов:

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

Шаблон событий:

Просто добавьте 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()

Мы добавили функциональность для удаления всех слушателей из объекта в класс Майклса и получили следующее:

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
 Eric Marcos01 апр. 2015 г., 01:10
@ Саймон прав, но вносит ошибку, потому что у нас могут быть несвязанные функции в обработчиках self .__. Исправить:self.__handlers = [h for h in self._handlers if getattr(h, 'im_self', False) != obj]
 Jonathan04 июл. 2011 г., 18:57
Недостатком использования этого является то, что вам необходимо сначала добавить событие, прежде чем зарегистрироваться в качестве подписчика. Если только издатели добавляют свои события (не обязательно, просто хорошая практика), то вы должны инициализировать издателей перед подписчиками, что является болью в крупных проекта
 Simon Bergot01 апр. 2013 г., 09:03
последний метод содержит ошибки, потому что обработчики self .__ изменяются во время итераций. Исправлено: `self .__ handlers = [h для h в self .__ handlers, если h.im_self! = Obj]`

Я нашел этот маленький скрипт на Ценные уроки. Кажется, у меня просто правильное соотношение простоты и мощности. Питер Тэтчер - автор следующего кода (лицензирование не упоминается).

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()
 florisla24 апр. 2013 г., 16:53
Использование set () вместо списка - это хорошо, чтобы обработчики не регистрировались дважды. Одним из следствий этого является то, что обработчики не вызываются в том порядке, в котором они были зарегистрированы. Не обязательно плохо, хотя ...
 Robino22 мар. 2018 г., 21:46
@ florisla может поменяться на OrderedSet, если того пожелает.

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"

Я создалEventManager класс (код в конце). Синтаксис следующий:

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

Вот пример:

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

Выход

Первоначальный салют
Привет, Оскар
Привет, Оскар

Теперь удалите приветствия
Привет, Оскар

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)

Модуль для рассмотрения. Это кажется приемлемым выбором для более требовательных приложений.

Py-notify - это пакет Python, предоставляющий инструменты для реализации шаблона программирования Observer. Эти инструменты включают сигналы, условия и переменные.

Signals - это списки обработчиков, которые вызываются при выдаче сигнала. Условия в основном являются булевыми переменными в сочетании с сигналом, который излучается при изменении состояния условия. Они могут быть объединены с использованием стандартных логических операторов (не, и т. Д.) В составные условия. Переменные, в отличие от условий, могут содержать любой объект Python, не только логические, но они не могут быть объединены.

buslane модуль.

Эта библиотека облегчает реализацию системы, основанной на сообщениях. Он поддерживает команды (один обработчик) и события (0 или несколько обработчиков). Buslane использует аннотации типа Python для правильной регистрации обработчика.

Простой пример:

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

Чтобы установить buslane, просто используйте pip:

$ pip install buslane

такие как объединение событий или повторить попытку, вы можете использовать шаблон Observable и зрелую библиотеку, которая реализует это.https: //github.com/ReactiveX/RxP. Observables очень распространены в Javascript и Java, и их очень удобно использовать для некоторых асинхронных задач.

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

ВЫХО:

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

который должен работать нормально. Что вам нужно сделать, это просто наследоватьObserver в классе, а затем использоватьobserve(event_name, callback_fn) чтобы прослушать конкретное событие. Всякий раз, когда это конкретное событие запускается где-нибудь в коде (то есть.Event('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')

Ваш ответ на вопрос