Method 1: creating your own class
You can implement this with a generator and by subclassing list. You will have to override every method you want to watch. Here is a simple example.
def watcher(name=''):
while True:
x = yield
msg = f'{name} was {x}'
y = yield
if y is not None:
msg += f', now is {y}'
print(msg)
class List(list):
def __init__(self, *args, gen=None, **kwargs):
self.__gen = gen
next(gen)
super().__init__(*args, **kwargs)
def __add__(self, other):
try:
self.__gen.send(self)
super().__add__(other)
self.__gen.send(self)
except:
next(self.__gen)
raise
def __setitem__(self, *args, **kwargs):
try:
self.__gen.send(self)
super().__setitem__(*args, **kwargs)
self.__gen.send(self)
except:
next(self.__gen)
raise
def append(self, value):
self.__gen.send(self)
super().append(value)
self.__gen.send(self)
Examples:
owl = watcher('Enemies')
enemies = List(["Moloch", "Twilight Lady", "Big Figure", "Captain Carnage", "Nixon"], gen=owl)
enemies.append('Alien')
# prints:
Enemies was ['Moloch', 'Twilight Lady', 'Big Figure', 'Captain Carnage', 'Nixon'], now is ['Moloch', 'Twilight Lady', 'Big Figure', 'Captain Carnage', 'Nixon', 'Alien']
enemies[-1] = 'Aliens'
# prints:
Enemies was ['Moloch', 'Twilight Lady', 'Big Figure', 'Captain Carnage', 'Nixon', 'Alien'], now is ['Moloch', 'Twilight Lady', 'Big Figure', 'Captain Carnage', 'Nixon', 'Aliens']
Method 2: Monkey-patching list
Here is an example that monkey-patches the list built-in type to add wrapped methods. In this case we are just adding capitalized versions of the methods that mutate the list; i.e. .append becomes .Append`.
Code for monkey-patching is from https://gist.github.com/bricef/1b0389ee89bd5b55113c7f3f3d6394ae. You can just copy this to a file called patch.py and use from patch import monkey_patch_list
import ctypes
from types import MappingProxyType, MethodType
# figure out side of _Py_ssize_t
if hasattr(ctypes.pythonapi, 'Py_InitModule4_64'):
_Py_ssize_t = ctypes.c_int64
else:
_Py_ssize_t = ctypes.c_int
# regular python
class _PyObject(ctypes.Structure):
pass
_PyObject._fields_ = [
('ob_refcnt', _Py_ssize_t),
('ob_type', ctypes.POINTER(_PyObject))
]
# python with trace
if object.__basicsize__ != ctypes.sizeof(_PyObject):
class _PyObject(ctypes.Structure):
pass
_PyObject._fields_ = [
('_ob_next', ctypes.POINTER(_PyObject)),
('_ob_prev', ctypes.POINTER(_PyObject)),
('ob_refcnt', _Py_ssize_t),
('ob_type', ctypes.POINTER(_PyObject))
]
class _DictProxy(_PyObject):
_fields_ = [('dict', ctypes.POINTER(_PyObject))]
def reveal_dict(proxy):
if not isinstance(proxy, MappingProxyType):
raise TypeError('dictproxy expected')
dp = _DictProxy.from_address(id(proxy))
ns = {}
ctypes.pythonapi.PyDict_SetItem(ctypes.py_object(ns),
ctypes.py_object(None),
dp.dict)
return ns[None]
def get_class_dict(cls):
d = getattr(cls, '__dict__', None)
if d is None:
raise TypeError('given class does not have a dictionary')
if isinstance(d, MappingProxyType):
return reveal_dict(d)
return d
class Listener:
def __init__(self):
self._g = None
def __call__(self, x=None):
if x is None:
return self._g
self._g = x
def send(self, val):
if self._g:
self._g.send(val)
def monkey_patch_list(decorator, mutators=None):
if not mutators:
mutators = (
'append', 'clear', 'extend', 'insert', 'pop', 'remove',
'reverse', 'sort'
)
d_list = get_class_dict(list)
d_list['_listener'] = Listener()
for m in mutators:
d_list[m.capitalize()] = decorator(d_list.get(m))
Now we can basically use what we had above, which is defining a decorator that wraps the list methods and yeilds the list's str representation before mutation and again after mutation. You can then pass any function that accepts two arguments and a name keyword argument to handle the alerting.
def before_after(clsm):
'''decorator for list class methods'''
def wrapper(self, *args, **kwargs):
self._listener.send(self)
out = clsm(self, *args, **kwargs)
self._listener.send(self)
return out
return wrapper
class Watchman:
def __init__(self):
self.guarding = []
def watch(self, lst, fn, name='list'):
self.guarding.append((lst, name))
w = self._watcher(fn, name)
lst._listener(w)
@staticmethod
def _watcher(fn, name):
def gen():
while True:
x = yield
x = str(x)
y = yield
y = str(y)
print(fn(x, y, name=name))
g = gen()
next(g)
return g
Now you can patch and use the new methods with the standard list constructor.
def enemies_changed(old, new, name='list'):
print(f"{name} was {old}, now are {new}")
# update the list methods with the wrapper
monkey_patch_list(before_after)
enemies = ["Moloch", "Twilight Lady", "Big Figure", "Captain Carnage", "Nixon"]
owl = Watchman()
owl.watch(enemies, enemies_changed, 'Enemies')
enemies.Append('Alien')
# prints:
Enemies was ['Moloch', 'Twilight Lady', 'Big Figure', 'Captain Carnage', 'Nixon'],
now are ['Moloch', 'Twilight Lady', 'Big Figure', 'Captain Carnage', 'Nixon', 'Alien']
Method 3: overloading the list built-in with your own implementation
This method is similar to method 1, but you can still use the list constructor as usual. We are basically going to over-write the built-in list class with our own subclassed version. The methods are the same syntax, we are just adding a listener and receiver attribute, we are also wrapping the mutating methods so that the listener picks up on them and sends a signal to the receiver is one is set.
# save the built-in list
_list = list
class list(_list):
def __init__(self, *args, emit_change=False, **kwargs):
super().__init__(*args, **kwargs)
self._emit = emit_change
self._listener = self._make_gen() if emit_change else None
self._init_change_emitter()
self._receiver = None
@property
def emit_change(self):
return self._emit
@property
def emitter(self):
return self._emitter
def _make_gen(self):
def gen():
while True:
x = yield
x = str(x)
y = yield
y = str(y)
yield (x, y)
g = gen()
next(g)
return g
def _init_change_emitter(self):
def before_after(clsm):
def wrapper(*args, **kwargs):
if self._listener:
self._listener.send(self)
out = clsm(*args, **kwargs)
before, after = self._listener.send(self)
if self._receiver:
self._receiver.send((before, after))
else:
out = clsm(*args, **kwargs)
return out
return wrapper
mutators = (
'append', 'clear', 'extend', 'insert', 'pop', 'remove',
'reverse', 'sort'
)
for m_str in mutators:
m = self.__getattribute__(m_str)
self.__setattr__(m_str, before_after(m))
At this point list works just like before. Using list('abc') returns the output ['a', 'b', 'c'], as you would expect, and so does list('abc', emit_change=True). The additional key-word argument allows to hook in receiver that gets sent before and after snippets of the list when it mutates.
A list constructed using brackets will need to be passed through the list constructor to turn on listening/emitting.
Example:
class Watchman:
def __init__(self):
self.guarding = []
def watch(self, lst, fn, name='list'):
if not lst._listener:
raise ValueError(
'Can only watch lists initialized with `emit_change=True`.'
)
r = self._make_receiver(fn, name)
lst._receiver = r
self.guarding.append((lst, name, r))
def _make_receiver(self, fn, name):
def receiver():
while True:
x, y = yield
print(fn(x, y, name=name))
r = receiver()
next(r)
return r
def enemies_changed(old, new, name):
return f"{name} was {old}\nNow is {new}"
enemies = ["Moloch", "Twilight Lady", "Big Figure", "Captain Carnage", "Nixon"]
enemies = list(enemies, emit_change=True)
owl = Watchman()
owl.watch(enemies, enemies_changed, 'Enemies')
enemies.append('Alien')
# prints:
Enemies was: ['Moloch', 'Twilight Lady', 'Big Figure', 'Captain Carnage', 'Nixon']
Now is: ['Moloch', 'Twilight Lady', 'Big Figure', 'Captain Carnage', 'Nixon', 'Alien']
Hope one of these helps!
