In peewee, I need to write some hook functions for each database operation, so that when the database content is changed, the corresponding hook function is called. This should be a very simple method overload operation: by method overload, the hook function is called before calling the parent class method. But I tried several methods and they didn't work. I'm not sure if overloading is the best choice to achieve my idea.
Here is my code:
class MyModel(pw.Model):
signal_callables = {}
support_actions = (
"delete",
"update",
"insert",
"insert_from",
"replace",
"replace_many",
"insert_many",
"delete_by_id",
"delete_instance",
"save",
"set_by_id",
)
calling = defaultdict()
# def __init__(self, *args, **kwargs):
# super().__init__(*args, **kwargs)
@classmethod
def signals(
cls, operates: Union[str, List], func: callable, exec_delay_ms: int = 0
):
if isinstance(operates, str):
cls._reg_signals(operates, func, exec_delay_ms)
elif isinstance(operates, list):
if not operates:
operates = cls.support_actions
for operate in operates:
cls._reg_signals(operate, func, exec_delay_ms)
else:
raise ValueError("Operate must be a str or a list of str")
@classmethod
def _reg_signals(cls, operate: str, func: callable, exec_delay_ms: int):
assert operate in cls.support_actions, "Operate must in "
# assert isinstance(model, pw.Model), "Model must be a subclass of peewee.Model"
assert callable(func), "Func must be a callable"
if not cls.signal_callables.get(operate):
cls.signal_callables[operate] = {}
model_name = cls.__name__
assert model_name != "Model", "Base model can't register signal"
if not cls.signal_callables[operate].get(model_name):
cls.signal_callables[operate][model_name] = []
cls.signal_callables[operate][model_name].append((func, exec_delay_ms))
@classmethod
def delete(cls):
cls._call_signal("delete")
return super().delete()
@classmethod
def update(cls, *argc, **kwargs):
cls._call_signal("update")
return super().update(*argc, **kwargs)
@classmethod
def save(cls, *argc, **kwargs):
cls._call_signal("save")
return super().save(*argc, **kwargs)
...
But it throw exceptions when I save data:
File "/usr/local/python3/lib/python3.7/site-packages/peewee.py", line 6620, in create
inst.save(force_insert=True)
│ └ <classmethod object at 0xffff85a12510>
└ <LogCollect: None>
File "/root/lz/xxx/utils/database/db.py", line 147, in save
return super().save(*argc, **kwargs)
│ └ {'force_insert': True}
└ ()
TypeError: save() missing 1 required positional argument: 'self'
It also don't work, if I rewrite in this way:
@classmethod
def save(cls, *argc, **kwargs):
cls._call_signal("save")
return pw.Model.save(*argc, **kwargs)
Here is the error info:
File "/root/lz/xxx/utils/database/db.py", line 147, in save
return pw.Model.save(*argc, **kwargs)
│ │ │ │ └ {'force_insert': True}
│ │ │ └ ()
│ │ └ <function Model.save at 0xffff85a5bf80>
│ └ <Model: Model>
└ <module 'peewee' from '/usr/local/python3/lib/python3.7/site-packages/peewee.py'>
TypeError: super(type, obj): obj must be an instance or subtype of type