Tortoise-orm信号实现及使用场景源码详解

# -*- coding: utf-8 -*-
"""
This example demonstrates model signals usage
"""
from typing import List, Optional, Type
from tortoise import BaseDBAsyncClient, Tortoise, fields, run_async
from tortoise.models import Model
from tortoise.signals import post_delete, post_save, pre_delete, pre_save
class Signal(Model):
    id = fields.IntField(pk=True)
    name = fields.TextField()
    class Meta:
        table = "signal"
    def __str__(self):
        return self.name
@pre_save(Signal)
async def signal_pre_save(
    sender: "Type[Signal]", instance: Signal, using_db, update_fields
) -> None:
    print('signal_pre_save', sender, instance, using_db, update_fields)
@post_save(Signal)
async def signal_post_save(
    sender: "Type[Signal]",
    instance: Signal,
    created: bool,
    using_db: "Optional[BaseDBAsyncClient]",
    update_fields: List[str],
) -> None:
    print('post_save', sender, instance, using_db, created, update_fields)
@pre_delete(Signal)
async def signal_pre_delete(
    sender: "Type[Signal]", instance: Signal, using_db: "Optional[BaseDBAsyncClient]"
) -> None:
    print('pre_delete', sender, instance, using_db)
@post_delete(Signal)
async def signal_post_delete(
    sender: "Type[Signal]", instance: Signal, using_db: "Optional[BaseDBAsyncClient]"
) -> None:
    print('post_delete', sender, instance, using_db)
async def run():
    await Tortoise.init(db_url="sqlite://:memory:", modules={"models": ["__main__"]})
    await Tortoise.generate_schemas()
    # pre_save,post_save will be send
    signal = await Signal.create(name="Signal")
    signal.name = "Signal_Save"
    # pre_save,post_save will be send
    await signal.save(update_fields=["name"])
    # pre_delete,post_delete will be send
    await signal.delete()
if __name__ == "__main__":
    run_async(run())

from enum import Enum
from typing import Callable
Signals = Enum("Signals", ["pre_save", "post_save", "pre_delete", "post_delete"])
def post_save(*senders) -> Callable:
    """
    ReGISter given models post_save signal.
    :param senders: Model class
    """
    def decorator(f):
        for sender in senders:
            sender.register_listener(Signals.post_save, f)
        return f
    return decorator
def pre_save(*senders) -> Callable:
    ...
def pre_delete(*senders) -> Callable:
    ...
def post_delete(*senders) -> Callable:
    ...

# 给模型Signal注册一个监听者，它是方法signal_pre_save
@pre_save(Signal)
async def signal_pre_save(
    sender: "Type[Signal]", instance: Signal, using_db, update_fields
) -> None:
    print('signal_pre_save', sender, instance, using_db, update_fields)

class Model:
    ...
    @claSSMethod
    def register_listener(cls, signal: Signals, listener: Callable):
        ...
        if not callable(listener):
            raise ConfigurationError("Signal listener must be callable!")
        # 检测是否已经注册过
        cls_listeners = cls._listeners.get(signal).setdefault(cls, [])  # type:ignore
        if listener not in cls_listeners:
            # 注册监听者
            cls_listeners.append(listener)

    async def save(
        self,
        using_db: Optional[BaseDBAsyncClient] = None,
        update_fields: Optional[Iterable[str]] = None,
        force_create: bool = False,
        force_update: bool = False,
    ) -> None:
        ...
        # 执行保存前的信号
        await self._pre_save(db, update_fields)
        if force_create:
            await executor.execute_insert(self)
            created = True
        elif force_update:
            rows = await executor.execute_update(self, update_fields)
            if rows == 0:
                raise IntegrityError(f"Can't update object that doesn't exist. PK: {self.pk}")
            created = False
        else:
            if self._saved_in_db or update_fields:
                if self.pk is None:
                    await executor.execute_insert(self)
                    created = True
                else:
                    await executor.execute_update(self, update_fields)
                    created = False
            else:
                # TODO: Do a merge/upsert operation here instead. Let the executor determine an optimal strategy for each DB engine.
                await executor.execute_insert(self)
                created = True
        self._saved_in_db = True
        # 执行保存后的信号
        await self._post_save(db, created, update_fields)

# -*- coding: utf-8 -*-
from enum import Enum
from typing import Callable, Dict
# 声明枚举信号量
Signals = Enum("Signals", ["before_process", "after_process"])
# 处理前的装饰器
def before_process(*senders):
    def decorator(f):
        for sender in senders:
            sender.register_listener(Signals.before_process, f)
        return f
    return decorator
# 处理后的装饰器
def after_process(*senders):
    def decorator(f):
        for sender in senders:
            sender.register_listener(Signals.after_process, f)
        return f
    return decorator
class Model(object):
    _listeners: Dict = {
        Signals.before_process: {},
        Signals.after_process: {}
    }
    @classmethod
    def register_listener(cls, signal: Signals, listener: Callable):
        """注册监听者"""
        # 判断是否已经存在监听者
        cls_listeners = cls._listeners.get(signal).setdefault(cls, [])
        if listener not in cls_listeners:
            # 如果不存在，则添加监听者
            cls_listeners.append(listener)
    def _before_process(self):
        # 取出before_process监听者
        cls_listeners = self._listeners.get(Signals.before_process, {}).get(self.__class__, [])
        for listener in cls_listeners:
            # 调用监听者
            listener(self.__class__, self)
    def _after_process(self):
        # 取出after_process监听者
        cls_listeners = self._listeners.get(Signals.after_process, {}).get(self.__class__, [])
        for listener in cls_listeners:
            # 调用监听者
            listener(self.__class__, self)
class SignalModel(Model):
    def process(self):
        """真正的调用端"""
        self._before_process()
        print("Processing")
        self._after_process()
# 注册before_process信号
@before_process(SignalModel)
def before_process_listener(*args, **kwargs):
    print("before_process_listener1", args, kwargs)
# 注册before_process信号
@before_process(SignalModel)
def before_process_listener(*args, **kwargs):
    print("before_process_listener2", args, kwargs)
# 注册after_process信号
@after_process(SignalModel)
def before_process_listener(*args, **kwargs):
    print("after_process_listener", args, kwargs)
if __name__ == '__main__':
    sm = SignalModel()
    sm.process()