I have written a simple event emitter in Python, which allows subscribing to events and emitting events (along with monitoring events too). I'm rather new to Python, and would like some feedback on the overall style of my code (I tried to follow PEP 0008) as well as how it works.
import inspect
_DEFAULT = object()
class Event(object):
def __init__(self, name, properties={}, cancellable=False, read_only=False,
monitor=False):
self._name = name
self._properties = properties
self._cancellable = cancellable
self._read_only = read_only
self._monitor = monitor
self._cancelled = False
@property
def name(self):
return self._name
def get(self, key, default=_DEFAULT):
if key not in self and default == _DEFAULT:
raise KeyError(key)
return self._properties[key] if key in self else default
@property
def cancellable(self):
return self._cancellable
@property
def cancelled(self):
return self._cancelled
@property
def read_only(self):
return self._read_only
@property
def monitor(self):
return self._monitor
def _enforce_read_only(self):
if self._read_only:
raise RuntimeError('event is read only')
if self._monitor:
raise RuntimeError('event is monitor only')
def set(self, key, value):
self._enforce_read_only()
if key not in self:
raise KeyError(key)
self._properties[key] = value
@cancelled.setter
def cancelled(self, cancelled):
self._enforce_read_only()
if not self.cancellable:
raise RuntimeError('event is not cancellable')
self._cancelled = cancelled
def cancel(self):
self.cancelled = True
def make_read_only(self):
return Event(self._name, self._properties, self._cancellable, True,
self._monitor)
def make_monitor(self):
return Event(self._name, self._properties, self._cancellable,
self._read_only, True)
def __contains__(self, key):
return key in self._properties
class EventEmitter(object):
def __init__(self):
self._monitors = {}
self._listeners = {}
def _emit(self, listeners, event):
name = str(event.name)
if name in listeners:
for listener in listeners[name]:
listener(event)
def emit_event(self, event):
self._emit(self._monitors, event.make_monitor())
self._emit(self._listeners, event)
return event
def emit(self, name, properties={}, cancellable=False, read_only=False,
monitor=False):
return self.emit_event(Event(str(name), properties, cancellable,
read_only, monitor))
def _check_func(self, func):
argspec = inspect.getargspec(func)
argcount = len(argspec.args)
bound = hasattr(func, '__self__') and func.__self__ is not None
if bound:
if argcount != 2:
raise TypeError('func {0} must accept 2 arguments, not {1}'
.format(func.__name__, argcount))
elif argcount != 1:
raise TypeError('func {0} must accept 1 argument, not {1}'.format(
func.__name__, argcount))
def _listen(self, listeners, name, func):
self._check_func(func)
name = str(name)
if name not in listeners:
listeners[name] = []
listeners[name].append(func)
def monitor(self, name, *funcs):
for func in funcs:
self._listen(self._monitors, name, func)
def on(self, name, *funcs):
for func in funcs:
self._listen(self._listeners, name, func)
There are a few things I personally don't feel comfortable with in this code, however I am not sure of the best ways to tackle them.
EventEmitter._check_func
looks a bit hacky to me, but I'm not sure if there's a better way to handle bound methods.EventEmitter.emit
: Would it be better converting name to a string insideEvent.__init__
so that we can be sure it's a string? Because currently it's converted to a string in bothemit
and_emit
(stringified in_emit
in case an event is provided toemit_event
that does not have a name that is a string).Event.make_read_only
/Event.make_monitor
: I don't like how I have to explicitly provide very variable from the current event to the new one, as I have to update both methods if the signature ofEvent.__init__
changes.