脚本之家，脚本语言编程技术及教程分享平台！
分类导航

Python|VBS|Ruby|Lua|perl|VBA|Golang|PowerShell|Erlang|autoit|Dos|bat|shell|

服务器之家 - 脚本之家 - Python - Tortoise-orm信号实现及使用场景源码详解

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

2023-05-29 16:36it_miclon Python

这篇文章主要为大家介绍了Tortoise-orm信号实现及使用场景源码详解，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪

场景

在使用Tortoise操作数据库的时候发现，通过对操作数据库模型加以装饰器，如@pre_save(Model)，可以实现对这个模型在savue时，自动调用被装饰的方法，从而实现对模型的一些操作。

在此先从官方文档入手，看一下官方的对于模型信号的Example

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
# -*- 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())

以上代码可直接复制后运行，运行后的结果：

signal_pre_save <class '__main__.Signal'> Signal <tortoise.backends.sqlite.client.SqliteClient object at 0x7f8518319400> None
post_save <class '__main__.Signal'> Signal <tortoise.backends.sqlite.client.SqliteClient object at 0x7f8518319400> True None
signal_pre_save <class '__main__.Signal'> Signal_Save <tortoise.backends.sqlite.client.SqliteClient object at 0x7f8518319400> ['name']
post_save <class '__main__.Signal'> Signal_Save <tortoise.backends.sqlite.client.SqliteClient object at 0x7f8518319400> False ['name']
pre_delete <class '__main__.Signal'> Signal_Save <tortoise.backends.sqlite.client.SqliteClient object at 0x7f8518319400>
post_delete <class '__main__.Signal'> Signal_Save <tortoise.backends.sqlite.client.SqliteClient object at 0x7f8518319400>

可以发现，对模型进行保存和删除时候，都会调用对应的信号方法。

源码

从导包可以得知，tortoise的所有信号方法都在tortoise.signals中。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
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:
    ...

其内部实现的四个信号方法分别是模型的保存后，保存前，删除前，删除后。

其内部装饰器代码也十分简单，就是对装饰器中的参数（也就是模型），注册一个监听者，而这个监听者，其实就是被装饰的方法。

如上面的官方示例中：

?
1
2
3
4
5
6
# 给模型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)

而到了Model类中，自然就有一个register_listener方法，定睛一看，上面示例Signal中并没有register_listener方法，所以自然就想到了，这个方法必定在父类Model中。

?
1
2
3
4
5
6
7
8
9
10
11
12
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)

接下来注册后，这个listeners就会一直跟着这个Signal类。只需要在需要操作关键代码的地方，进行调用即可。

看看在模型save的时候，都干了什么？

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
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)

抛开其他代码，可以看到，在模型save的时候，其实是先执行保存前的信号，然后执行保存后的信号。

自己实现一个信号

有了以上的经验，可以自己实现一个信号，比如我打算做个数据处理器的类，我想在这个处理器工作中，监听处理前/后的信号。

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
# -*- 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()

输出结果：

before_process_listener1 (<class '__main__.SignalModel'>, <__main__.SignalModel object at 0x7ff700116e50>) {}
before_process_listener2 (<class '__main__.SignalModel'>, <__main__.SignalModel object at 0x7ff700116e50>) {}
Processing
after_process_listener (<class '__main__.SignalModel'>, <__main__.SignalModel object at 0x7ff700116e50>) {}

总结

笔者通过对`tortoise-orm`源码的学习，抽丝剥茧，提取了信号实现的方式。其核心就是通过一个字典存储调用方自定义的process方法，然后分别在真正的调用端的前/后触发这些自定义方法即可。

以上就是Tortoise-orm信号实现及使用场景源码详解的详细内容，更多关于Tortoise orm信号场景的资料请关注服务器之家其它相关文章！

原文链接：https://juejin.cn/post/7177652667911667768

  • 使用场景
  • 信号
  • Tortoise-orm

    • 延伸 · 阅读

    精彩推荐
    • PythonOpenCV-Python实现腐蚀与膨胀的实例

      OpenCV-Python实现腐蚀与膨胀的实例

      形态学操作主要包含：腐蚀，膨胀，开运算，闭运算，形态学梯度运算，顶帽运算，黑帽运算等操作，本文主要介绍了腐蚀与膨胀，感兴趣的小伙伴们可以...

      一天一篇Python库9772021-11-30
    • PythonPython 并行加速技巧分享

      Python 并行加速技巧分享

      这篇文章主要介绍了Python 并行加速技巧分享，文章围绕文章主题展开详细的内容介绍，具有一定的参考价值，需要的小伙伴可以参考一下...

      Python数据挖掘5632023-02-09
    • Python手把手教你如何安装Pycharm(详细图文教程)

      手把手教你如何安装Pycharm(详细图文教程)

      这篇文章主要介绍了手把手教你如何安装Pycharm(详细图文教程)，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧...

      dcpeng8712021-04-22
    • PythonPyQt5实现QLineEdit正则表达式输入验证器

      PyQt5实现QLineEdit正则表达式输入验证器

      这篇文章主要介绍了PyQt5实现QLineEdit正则表达式输入验证器，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的...

      鱼子酱1266992021-10-04
    • PythonPython处理文本文件中控制字符的方法

      Python处理文本文件中控制字符的方法

      最近在使用Python的时候遇到过文档中出现控制字符报错的问题。想着总结一下，方便以后需要或这同样遇到问题的朋友，下面这篇文章主要介绍了Python处理...

      Python教程网5482020-09-20
    • PythonPython实现SMTP邮件发送

      Python实现SMTP邮件发送

      这篇文章主要介绍了基于python实现SMTP发送邮件教程,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参...

      追风的小蚂蚁12342020-06-17
    • PythonPycharm2017版本设置启动时默认自动打开项目的方法

      Pycharm2017版本设置启动时默认自动打开项目的方法

      今天小编就为大家分享一篇Pycharm2017版本设置启动时默认自动打开项目的方法，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧...

      Vimiix8822021-04-14
    • PythonTensorFlow在MAC环境下的安装及环境搭建

      TensorFlow在MAC环境下的安装及环境搭建

      小编在论坛中看到很多朋友在寻找TensorFlow的环境搭建图文步骤以及安装的具体流程，在此小编给大家整理了一篇非常详细的图文流程，希望能够帮助到你。...

      狼.wrz8542020-12-17
    最近更新
    编辑推荐
    阅读排行
    热门标签
    2048 　 6006 　 6174 　 SLOTS 　 MELIAE 　 FEEDPARSER 　 暂存 　 持久性管理 　 pickle模块 　 THREADING 　 reduce 　 nignx 　 装饰器 　 函数式编程 　 兔子毒药 　 contextlib 　 监控文件 　 流量监控 　 Web框架 　 PIL库 　 高斯模糊 　 wxPython 　 Tortoise-orm 　 自定义模块 　 Python开发 　 元类 　 metaclass 　 魔术方法 　 DNSPod 　 动态解析域名 　
    © 2019-2023 脚本之家 | 服务器之家(www.tuohang.net)旗下站点 版权所有关于我们联系我们版权申明网站地图