-
Notifications
You must be signed in to change notification settings - Fork 1
/
observe.py
66 lines (58 loc) · 2.17 KB
/
observe.py
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
class Event:
"""
Object sent by the Observable.notify() call
This will be passed to callables being notified
"""
def __init__(self, name, source, kw):
"""
:param name: The event being fired
:param source: the source of the event (Observable object)
:param kw: a dictionary to attach data
"""
self.name = name
self.source = source
self.kwargs = kw
def __str__(self):
return self.name
class Observable:
def __init__(self):
# dicts of form {eventname: [callbacks]}
self.callbacks = {}
def remove_event(self, eventname):
"""
Removes an event
Will just do nothing if event is not already a key in callbacks
:param eventname: the name of the vent to remove
:return:
"""
self.callbacks.pop(eventname, None)
def subscribe(self, eventname, callback):
"""
Add a callback to an event
:param eventname: the event name to add a callback for (string)
:param callback: a callable to be run when notify() is called, must take only one required argument
"""
if eventname not in self.callbacks.keys():
self.callbacks[eventname] = [callback]
else:
self.callbacks[eventname].append(callback)
def unsubscribe(self, eventname, callback):
"""
Unsubscribe a callback from an event
NB: If a callback has been subscribed twice, only one instance will be removed
:param eventname: the event to remove from (string)
:param callback: the callable to remove
"""
if eventname in self.callbacks.keys():
if callback in self.callbacks[eventname]:
self.callbacks[eventname].remove(callback)
def notify(self, eventname, **kwargs):
"""
Call all callbacks for the given events
:param eventname: the event to notify for (string)
:param kwargs: keyword args to pass with the event object
"""
event = Event(eventname, self, kwargs)
if event.name in self.callbacks.keys():
for fn in self.callbacks[event.name][:]:
fn(event)