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: {}}}}}