forked from tortoise/tortoise-orm
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathsignals.py
More file actions
69 lines (50 loc) · 1.69 KB
/
signals.py
File metadata and controls
69 lines (50 loc) · 1.69 KB
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
66
67
68
69
"""
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(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(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(sender, instance, using_db)
@post_delete(Signal)
async def signal_post_delete(
sender: "Type[Signal]", instance: Signal, using_db: "Optional[BaseDBAsyncClient]"
) -> None:
print(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())