Mercurial > repos > shellac > guppy_basecaller
diff env/lib/python3.7/site-packages/rdflib/events.py @ 5:9b1c78e6ba9c draft default tip
"planemo upload commit 6c0a8142489327ece472c84e558c47da711a9142"
author | shellac |
---|---|
date | Mon, 01 Jun 2020 08:59:25 -0400 |
parents | 79f47841a781 |
children |
line wrap: on
line diff
--- a/env/lib/python3.7/site-packages/rdflib/events.py Thu May 14 16:47:39 2020 -0400 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,93 +0,0 @@ -__doc__ = """ -Dirt Simple Events - -A Dispatcher (or a subclass of Dispatcher) stores event handlers that -are 'fired' simple event objects when interesting things happen. - -Create a dispatcher: - - >>> d = Dispatcher() - -Now create a handler for the event and subscribe it to the dispatcher -to handle Event events. A handler is a simple function or method that -accepts the event as an argument: - - >>> def handler1(event): print(repr(event)) - >>> d.subscribe(Event, handler1) - -Now dispatch a new event into the dispatcher, and see handler1 get -fired: - - >>> d.dispatch(Event(foo='bar', data='yours', used_by='the event handlers')) - <rdflib.events.Event ['data', 'foo', 'used_by']> -""" - -__all__ = ['Event', 'Dispatcher'] - - -class Event(object): - """ - An event is a container for attributes. The source of an event - creates this object, or a subclass, gives it any kind of data that - the events handlers need to handle the event, and then calls - notify(event). - - The target of an event registers a function to handle the event it - is interested with subscribe(). When a sources calls - notify(event), each subscriber to that event will be called in no - particular order. - """ - - def __init__(self, **kw): - self.__dict__.update(kw) - - def __repr__(self): - attrs = list(self.__dict__.keys()) - attrs.sort() - return '<rdflib.events.Event %s>' % ([a for a in attrs],) - - -class Dispatcher(object): - """ - An object that can dispatch events to a privately managed group of - subscribers. - """ - - _dispatch_map = None - - def set_map(self, amap): - self._dispatch_map = amap - - def get_map(self): - return self._dispatch_map - - def subscribe(self, event_type, handler): - """ Subscribe the given handler to an event_type. Handlers - are called in the order they are subscribed. - """ - if self._dispatch_map is None: - self.set_map({}) - lst = self._dispatch_map.get(event_type, None) - if lst is None: - lst = [handler] - else: - lst.append(handler) - self._dispatch_map[event_type] = lst - - def dispatch(self, event): - """ Dispatch the given event to the subscribed handlers for - the event's type""" - if self._dispatch_map is not None: - lst = self._dispatch_map.get(type(event), None) - if lst is None: - raise ValueError("unknown event type: %s" % type(event)) - for l in lst: - l(event) - - -def test(): - import doctest - doctest.testmod() - -if __name__ == '__main__': - test()