今天小編給大家分享一下Tortoise orm信號實現及使用場景是什么的相關知識點，內容詳細，邏輯清晰，相信大部分人都還太了解這方面的知識，所以分享這篇文章給大家參考一下，希望大家閱讀完這篇文章后有所收獲，下面我們一起來了解一下吧。
在使用Tortoise操作數據庫的時候發現，通過對操作數據庫模型加以裝飾器，如@pre_save(Model)，可以實現對這個模型在savue時，自動調用被裝飾的方法，從而實現對模型的一些操作。
在此先從官方文檔入手，看一下官方的對于模型信號的Example
# -*- 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中。
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)而到了Model類中，自然就有一個register_listener方法，定睛一看，上面示例Signal中并沒有register_listener方法，所以自然就想到了，這個方法必定在父類Model中。
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類。只需要在需要操作關鍵代碼的地方，進行調用即可。
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的時候，其實是先執行保存前的信號，然后執行保存后的信號。
有了以上的經驗，可以自己實現一個信號，比如我打算做個數據處理器的類，我想在這個處理器工作中，監聽處理前/后的信號。
# -*- 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信號實現及使用場景是什么”這篇文章的所有內容，感謝各位的閱讀！相信大家閱讀完這篇文章都有很大的收獲，小編每天都會為大家更新不同的知識，如果還想學習更多的知識，請關注億速云行業資訊頻道。
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
