Source code for pydispatch.dispatch

import types

from pydispatch.utils import WeakMethodContainer, EmissionHoldLock
from pydispatch.properties import Property


[docs]class Event(object): """Holds references to event names and subscribed listeners This is used internally by :class:`Dispatcher`. """ __slots__ = ('name', 'listeners', 'emission_lock') def __init__(self, name): self.name = name self.listeners = WeakMethodContainer() self.emission_lock = EmissionHoldLock(self) def add_listener(self, callback): self.listeners.add_method(callback) def remove_listener(self, obj): if isinstance(obj, (types.MethodType, types.FunctionType)): self.listeners.del_method(obj) else: self.listeners.del_instance(obj)
[docs] def __call__(self, *args, **kwargs): """Dispatches the event to listeners Called by :meth:`~Dispatcher.emit` """ if self.emission_lock.held: self.emission_lock.last_event = (args, kwargs) return for m in self.listeners.iter_methods(): r = m(*args, **kwargs) if r is False: return r
def __repr__(self): return '<{}: {}>'.format(self.__class__, self) def __str__(self): return self.name
[docs]class Dispatcher(object): """Core class used to enable all functionality in the library Interfaces with :class:`Event` and :class:`~pydispatch.properties.Property` objects upon instance creation. Events can be created by calling :meth:`register_event` or by the subclass definition:: class Foo(Dispatcher): _events_ = ['awesome_event', 'on_less_awesome_event'] Once defined, an event can be dispatched to listeners by calling :meth:`emit`. """ def __new__(cls, *args, **kwargs): def iter_bases(_cls): if _cls is object: raise StopIteration yield _cls for b in _cls.__bases__: for _cls_ in iter_bases(b): yield _cls_ props = {} events = set() for _cls in iter_bases(cls): for attr in dir(_cls): prop = getattr(_cls, attr) if attr not in props and isinstance(prop, Property): props[attr] = prop prop.name = attr _events = getattr(_cls, '_events_', []) events |= set(_events) cls._PROPERTIES_ = props cls._EVENTS_ = events obj = super(Dispatcher, cls).__new__(cls) obj._Dispatcher__init_events() return obj def __init__(self, *args, **kwargs): # Everything is handled by __new__ # This is only here to prevent exceptions being raised pass def __init_events(self): if hasattr(self, '_Dispatcher__events'): return self.__events = {} for name in self._EVENTS_: self.__events[name] = Event(name) self.__property_events = {} for name, prop in self._PROPERTIES_.items(): self.__property_events[name] = Event(name) prop._add_instance(self)
[docs] def register_event(self, *names): """Registers new events after instance creation Args: *names (str): Name or names of the events to register """ for name in names: if name in self.__events: continue self.__events[name] = Event(name)
[docs] def bind(self, **kwargs): """Subscribes to events or to :class:`~pydispatch.property.Property` updates Keyword arguments are used with the Event or Property names as keys and the callbacks as values:: class Foo(Dispatcher): name = Property() foo = Foo() foo.bind(name=my_listener.on_foo_name_changed) foo.bind(name=other_listener.on_name, value=other_listener.on_value) The callbacks are stored as weak references and their order is not maintained relative to the order of binding. """ props = self.__property_events events = self.__events for name, cb in kwargs.items(): if name in props: e = props[name] else: e = events[name] e.add_listener(cb)
[docs] def unbind(self, *args): """Unsubscribes from events or :class:`Property` updates Multiple arguments can be given. Each of which can be either the method that was used for the original call to :meth:`bind` or an instance object. If an instance of an object is supplied, any previously bound Events and Properties will be 'unbound'. """ props = self.__property_events.values() events = self.__events.values() for arg in args: for prop in props: prop.remove_listener(arg) for e in events: e.remove_listener(arg)
[docs] def emit(self, name, *args, **kwargs): """Dispatches an event to any subscribed listeners Note: If a listener returns ``False``, the event will stop dispatching to other listeners. Any other return value is ignored. Args: name (str): The name of the :class:`Event` to dispatch *args (Optional): Positional arguments to be sent to listeners **kwargs (Optional): Keyword arguments to be sent to listeners """ e = self.__property_events.get(name) if e is None: e = self.__events[name] return e(*args, **kwargs)
[docs] def emission_lock(self, name): """Holds emission of events and dispatches the last event on release The context manager returned will store the last event data called by :meth:`emit` and prevent callbacks until it exits. On exit, it will dispatch the last event captured (if any):: class Foo(Dispatcher): _events_ = ['my_event'] def on_my_event(value): print(value) foo = Foo() foo.bind(my_event=on_my_event) with foo.emission_lock('my_event'): foo.emit('my_event', 1) foo.emit('my_event', 2) >>> 2 Args: name (str): The name of the :class:`Event` or :class:`Property` Returns: A context manager to be used by the ``with`` statement. If available, this will also be an async context manager to be used with the ``async with`` statement (see `PEP 492`_). Note: The context manager is re-entrant, meaning that multiple calls to this method within nested context scopes are possible. .. _PEP 492: https://www.python.org/dev/peps/pep-0492/#asynchronous-context-managers-and-async-with """ e = self.__property_events.get(name) if e is None: e = self.__events[name] return e.emission_lock