How to override Model class in peewee?

36 Views Asked by At

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
0

There are 0 best solutions below