Source code for tyuio

'''
Tyuio is a Simple Event Dispatcher api

@author: Moises P. Sena <moisespsena@gmail.com>
'''


[docs]class EventType(object): ''' The Event Type ''' def __init__(self, evcls, evtype): self.evcls = evcls self.evtype = evtype def __hash__(self): return hash((self.evcls, self.evtype)) def __repr__(self): return "<EventType(%r, %r)>" % (self.evcls, self.evtype) def __str__(self): return "%r.%s" % (self.evcls, self.evtype) def __eq__(self, other): if isinstance(other, EventType): if other.evcls == self.evcls: if other.evtype == self.evtype: return True return False
[docs]class Event(object): """ Generic Event object to use with EventDispatcher. Example: >>> dispatcher = EventDispatcher() >>> def callback(e, *args, **kwargs): ... print(e.type) >>> dispatcher.add_event_listener("my_type", callback) <EventDispatcher()> >>> dispatcher.dispatch(Event("my_type")) my_type <Event('my_type')> """ def __init__(self, event_type): """ The constructor accepts an event type as string and a custom data """ self.type = event_type self.targets = [] self.called_listeners = None self._tmp_target = None def __enter__(self): """ Enter in new dispatch context Enter in new dispatch context. Append current target in ``self.targets`` list. :return: The target object Example: >>> class D1(EventDispatcher): pass >>> class D2(EventDispatcher): pass >>> d1 = D1() >>> d2 = D2() >>> event = Event("my_type") >>> with event(d1): ... print(event.source_target) ... print(event.current_target) ... print(event.targets) ... ... with event(d2): ... print(event.source_target) ... print(event.current_target) ... print(event.targets) <D1()> <D1()> [<D1()>] <D1()> <D2()> [<D1()>, <D2()>] """ assert self._tmp_target self.targets.append(self._tmp_target) self._tmp_target = None def __exit__(self, exc_type, exc_val, exc_tb): """ Return last inserted target object in ``self.targets`` list. :return: The target object """ assert not self._tmp_target self.targets.pop() def __call__(self, target): """ Add ``target`` for use it in ``with`` expression. :param target: The new event target :return: self event object """ assert not self._tmp_target self._tmp_target = target return self @property
[docs] def current_target(self): """ Return last target object in ``self.targets`` list. :return: The target object """ return self.targets[-1]
@property
[docs] def source_target(self): """ Return first target object in ``self.targets`` list. :return: The target object """ return self.targets[0]
def __repr__(self): return "<Event(%r)>" % (self.type)
[docs]class Listener(object): """Listener interface""" def __call__(self, event, *args, **kwargs): """Listener invoker""" raise NotImplementedError()
[docs]def is_iter(obj): ''' Check if obj contains '__iter__' attribute :param obj: the object of check :return: ``True`` if contains, other else, ``False`` ''' return hasattr(obj, '__iter__')
[docs]class EventDispatcher(object): """ Generic event dispatcher which listen and dispatch events Examples: >>> d = EventDispatcher() or >>> class Target(object): ... def __repr__(self): ... return '<%s object>' % self.__class__.__name__ >>> d = EventDispatcher(Target()) >>> d.target <Target object> Add listeners: >>> def callback(e, *args, **kwargs): ... pass >>> d.add_event_listener('my_event', callback) <EventDispatcher(target=<Target object>)> >>> d.add_event_listeners('my_event', callback, callback) <EventDispatcher(target=<Target object>)> >>> d.add_events_listeners(('my_event', [callback, callback])) <EventDispatcher(target=<Target object>)> >>> d.add_events_listeners(('my_event', [callback, callback]), \ ('another_event', [callback, callback])) <EventDispatcher(target=<Target object>)> """ def __init__(self, target=None): """Kwargs: - target: The event target. Default is ``None`` """ self._event_listeners = dict() self._target = target @property
[docs] def target(self): return self._target or self
def __del__(self): """ Remove all listener references at destruction time """ self._event_listeners = None
[docs] def has_listener(self, event_type, listener): """ Check if dispatch contains the ``listener`` of ``event_type`` event type. :param event_type: The event type :param listener: The listener callback :return: ``True`` if contains, other else, ``False`` """ # Check for event type and for the listener if event_type in self._event_listeners.keys(): return listener in self._event_listeners[event_type] else: return False
def _invoke_listener(self, listener, event, *args, **kwargs): return listener(event, *args, **kwargs)
[docs] def dispatch(self, event, *args, **kwargs): """ Dispatch all listeners of ``event.type`` :param event: The event object :param args: Args to pass at listener function :param kwargs: Keyword args to pass at listener function :return: the ``event`` object >>> dispatcher = EventDispatcher() >>> def callback(e, *args, **kwargs): ... print(e.type) >>> dispatcher.add_event_listener("my_type", callback) <EventDispatcher()> >>> dispatcher.dispatch(Event("my_type")) my_type <Event('my_type')> """ event.called_listeners = [] # Dispatch the event to all the associated listeners if event.type in self._event_listeners: listeners = self._event_listeners[event.type] with event(self.target): for listener in listeners: event.called_listeners.append(listener) # call listener and stop if returns False if self._invoke_listener(listener, event, *args, **kwargs) is False: break return event
[docs] def add_event_listener(self, event_type, listener): """ Add one listener at event type :param event_type: The event type :param listener: The listener :return: the ``self`` object Example: >>> d = EventDispatcher() >>> def callback(e, *args, **kwargs): ... pass >>> d.add_event_listener('my_event', callback) <EventDispatcher()> """ # Add listener to the event type if not self.has_listener(event_type, listener): listeners = self._event_listeners.get(event_type, []) listeners.append(listener) self._event_listeners[event_type] = listeners return self
on = add_event_listener
[docs] def add_event_listeners(self, event_type, *listeners): """ Add one or more listeners at event type :param event_type: The event type :param listeners: The listener list :return: the ``self`` object Example: >>> d = EventDispatcher() >>> def callback(e, *args, **kwargs): ... pass >>> d.add_events_listeners('my_event', [callback, callback]) <EventDispatcher()> """ if len(listeners) == 1 and is_iter(listeners[0]): listeners = listeners[0] for _ in listeners: self.add_event_listener(event_type, _) return self
[docs] def add_events_listeners(self, *types_with_listeners): """ Add multiples listeners on varios events types. :param types_with_listeners: tuples of (event_type, listeners) :return: the ``self`` object Example: >>> d = EventDispatcher() >>> def callback(e, *args, **kwargs): ... pass >>> d.add_events_listeners(('my_event', [callback, callback]), ('another_event', [callback, callback])) <EventDispatcher()> """ if len(types_with_listeners) == 1 and is_iter(types_with_listeners[0]): types_with_listeners = types_with_listeners[0] for _ in types_with_listeners: self.add_event_listeners(*_) return self
[docs] def remove_event_listener(self, event_type, listener): """ Remove one listener at event type :param event_type: The event type :param listener: The listener :return: the ``self`` object Example: >>> d = EventDispatcher() >>> def callback(e, *args, **kwargs): ... pass >>> d.remove_event_listener('my_event', callback) <EventDispatcher()> """ # Remove the listener from the event type if self.has_listener(event_type, listener): listeners = self._event_listeners[event_type] if len(listeners) == 1: # Only this listener remains so remove the key del self._event_listeners[event_type] else: # Update listeners chain listeners.remove(listener) self._event_listeners[event_type] = listeners return self
[docs] def remove_event_listeners(self, event_type, *listeners): """ Remove one or more listeners at event type :param event_type: The event type :param listeners: The listener list :return: the ``self`` object Example: >>> d = EventDispatcher() >>> def callback(e, *args, **kwargs): ... pass >>> d.remove_events_listeners('my_event', [callback, callback]) <EventDispatcher()> """ if len(listeners) == 1 and is_iter(listeners[0]): listeners = listeners[0] for _ in listeners: self.remove_event_listener(event_type, _) return self
[docs] def remove_events_listeners(self, *types_with_listeners): """ Removes multiples listeners on varios events types. :param types_with_listeners: tuples of (event_type, listeners) :return: the ``self`` object Example: >>> d = EventDispatcher() >>> def callback(e, *args, **kwargs): ... pass >>> d.remove_events_listeners(('my_event', [callback, callback]),('another_event', [callback, callback])) <EventDispatcher()> """ if len(types_with_listeners) == 1 and \ is_iter(types_with_listeners[0]): types_with_listeners = types_with_listeners[0] for _ in types_with_listeners: self.remove_event_listeners(*_) return self
[docs] def remove_all_listeners(self): """ Remove all events listeners :return: the ``self`` object """ self._event_listeners = {} return self
def __repr__(self): return "<%s(%s)>" % (self.__class__.__name__, self._target and \ ('target=%r' % self._target) or '')