/usr/lib/python3/dist-packages/rdflib/events.py is in python3-rdflib 4.2.1-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 | __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()
|