Context: Elegant solution needed to monitor and save database changes during runtime. To minimze IO, the solution should be based on signals rather than polling and comparing against file content.
Solution: Use a callback to start a save timer following database alteration. Thus, I subclassed a dict (Observer) and overridden __setitem__ method to handle the addition of new keys. As the dict content is loaded from a JSON, all nested dict object must also be converted after update() is called.
Edit: it works! can it be further improved?
Solution: Use a callback to start a save timer following database alteration. Thus, I subclassed a dict (Observer) and overridden __setitem__ method to handle the addition of new keys. As the dict content is loaded from a JSON, all nested dict object must also be converted after update() is called.
Edit: it works! can it be further improved?
#!/usr/bin/python3
from typing import Callable
class Observer(dict):
def __init__(self, *args, callback: Callable):
super().__init__(*args)
self.cb = callback
def __setitem__(self, item: any, value: any):
""" Fires a callback function when a value is changed """
value = Observer(value, callback=self.cb) if isinstance(value, dict) else value
super().__setitem__(item, value)
if isinstance(value, dict):
self._convert(value)
self.cb()
def update(self, *args):
""" Converts dict objects after update """
super().update(*args)
self._convert(self)
def _convert(self, d: dict):
""" Recursively converts nested dict to Observer subclass """
for key, value in d.items():
if isinstance(value, dict):
d[key] = Observer(value, callback=self.cb)
self._convert(value)
class Database(Observer):
def __init__(self, d):
super().__init__(self, callback=lambda: None) # ##
self.update(d)
def demo():
db = Database({1: {"foo": ["bar"]}, 2: {3: {"bubu": 0}}})
db[4] = {5: {6: {7: {}}}}
print(f"{type(db)}\n")
print("update")
print(f"{type(db[1])=}")
print(f"{type(db[2])=}")
print(f"{type(db[2][3])=}")
print("\nsetitem")
print(f"{type(db[4])=}")
print(f"{type(db[4][5])=}")
print(f"{type(db[4][5][6])=}")
print(f"{type(db[4][5][6][7])=}")
print()
print(db)
demo()Output:Quote:<class '__main__.Database'>
update
type(db[1])=<class '__main__.Observer'>
type(db[2])=<class '__main__.Observer'>
type(db[2][3])=<class '__main__.Observer'>
setitem
type(db[4])=<class '__main__.Observer'>
type(db[4][5])=<class '__main__.Observer'>
type(db[4][5][6])=<class '__main__.Observer'>
type(db[4][5][6][7])=<class '__main__.Observer'>
{1: {'foo': ['bar']}, 2: {3: {'bubu': 0}}, 4: {5: {6: {7: {}}}}}
