/usr/share/pyshared/gpyconf/events.py is in python-gpyconf 0.2-1.
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 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 | # %FILEHEADER%
from collections import defaultdict
from functools import partial
from types import FunctionType
class InvalidEvent(Exception):
"""
Raised if a non-defined event should be registered at a strict-mode
event register.
"""
def __init__(self, event):
self.event = event
def __str__(self):
return self.event
class EventRegister(object):
"""
Very simple event handler. Listening functions can register themselves
and will be called when the signal they're listening for is emitted.
There's no restriction for listening to events, you can listen to any
event that might be never called if you want to.
Example:
>>> events = EventRegister()
>>> @events.start
... def on_start_do_this():
... print "hi there"
>>> @events.end
... def on_end_do_that(a, b):
... print a, b
>>> events.emit('foo') # nothing will happen 'cause nobody's listening
>>> events.emit('start')
hi there
>>> events.emit('end', 42, 'x')
42 x
To avoid typos registering your events, you can use the 'strict' mode
(subclass `EventRegister` and define an :attr:`__events__` list).
Using the strict mode, you have to define all allowed events in the
:attr:`__events__` attribute. Then, if someone wants to register an event
not defined in that :attr:`__events__`, an `InvalidEvent` exception will be
raised.
>>> class MyStrictEvents(EventRegister):
... __events__ = ('hi', 'there')
>>> sevents = MyStrictEvents()
>>> @sevents.hi
... def on_hi():
... print "Hi!"
>>> @sevents.some_undefined_event
... def never_called():
... pass
Traceback (most recent call last):
...
InvalidEvent: some_undefined_event
You can have lazy callbacks, too:
>>> events = EventRegister()
>>> @events.some_signal
... def called_first():
... print "First callback"
>>> @events.some_signal(lazy=True)
... def called_last():
... print "Last callback"
>>> @events.some_signal
... def called_second():
... print "Second callback"
>>> events.emit('some_signal')
First callback
Second callback
Last callback
"""
strict = False
initialized = False
def __init__(self):
if hasattr(self, '__events__'):
self.strict = True
self.__events__ = list(self.__events__)
self.events = defaultdict(list)
self.all_events_listener = []
def __getattr__(self, event):
if event == '__events__':
raise AttributeError(event)
def register_event(*args, **kwargs):
if args and isinstance(args[0], FunctionType):
self.register_event(event, args[0])
return args[0]
else:
def wrapper(func):
self.register_event(event, func, *args, **kwargs)
return func
return wrapper
return register_event
def register_event(self, event, callback, lazy=False):
"""
Register ``callback`` for ``event``. This is similar to ::
@myinstance.myevent
def callback(...):
...
where ``myevent`` is the value of the ``event`` attribute.
"""
callback.__dict__['lazy'] = lazy
if event == 'all':
self.all_events_listener.append(callback)
elif self.strict and event not in self.__events__:
raise InvalidEvent(event)
else:
self.events[event].append(callback)
def emit(self, event, *args, **kwargs):
"""
Emit ``event``. Calls all callbacks registered for this ``event`` and
all callbacks registered for *all* events
(passing ``*args`` and ``**kwargs`` as parameters).
Raises :exc:`InvalidEvent` if mode is strict and ``event`` is not
defined in :attr:`__events__`.
"""
if self.strict and event not in self.__events__:
raise InvalidEvent(event)
lazy_callbacks = []
if event in self.events:
for func in self.events[event]:
if func.__dict__['lazy']:
lazy_callbacks.append(func)
else:
func(*args, **kwargs)
for func in self.all_events_listener:
if func.lazy:
lazy_callbacks.append(partial(event, func))
else:
func(event, *args, **kwargs)
for func in lazy_callbacks:
func(*args, **kwargs)
class GEventRegister(EventRegister):
"""
Event register for the :class:`GSignals` class.
Automatically converts signal names with underscores ("foo_bar") to names
with hyphens ("foo-bar").
Same API as :class:`EventRegister`.
"""
def __init__(self, events=None):
if events is not None:
self.__events__ = events
EventRegister.__init__(self)
def __getattr__(self, event):
if event == '__events__':
raise AttributeError(event)
# GSignals uses hyphens, not underscores
return EventRegister.__getattr__(self, event.replace('_', '-'))
class GSignals(object):
"""
GObject/GSignals-compatible class mixin.
Connected callbacks always have to take a ``sender`` as first argument
(this is for GSignals compatibility reasons).
:attr:`events` attribute: The :class:`GEventRegister`.
"""
__gsignals__ = None
def __init__(self):
events = self.__events__
if self.__gsignals__ is None:
self.__gsignals__ = {}
else:
events += self.__gsignals__.keys()
# throw away gsignals parameter definitions (this is gobject-C-stuff)
self.events = GEventRegister(events)
def connect(self, signal, callback, lazy=False):
""" Connect ``callback`` to ``signal`` """
self.events.register_event(signal, callback, lazy)
def emit(self, signal, *args, **kwargs):
""" Emit ``signal`` """
self.events.emit(signal, self, *args, **kwargs)
def add_events(self, events):
""" Add a list of events to the allowed events """
self.events.__events__ += list(events)
def add_event(self, event):
""" Add a event to the allowed events """
self.events.__events__ += [event]
if __name__ == '__main__':
from doctest import testmod
testmod()
|