Tortoise ORM

# Tortoise ORM

# 介绍

Tortoise ORM是Python中一个专为异步开发设计的ORM框架，它可以用面向对象的方式操作数据库，同时充分利用Python异步编程的优势。

它的设计灵感来自Django ORM，因此和Django ORM的用法实际差不多。

# 特点

原生异步支持：所有操作都是async/await风格，不会阻塞事件循环。
Django风格API：对熟悉Django的开发者非常友好。
多数据库支持：支持SQLite、PostgreSQL、MySQL等主流数据库。
自动表结构生成：根据模型类自动生成数据库表。
类型安全：结合Python类型注解，减少数据错误。
高效查询：支持链式查询、复杂条件过滤、聚合操作等。

# 字段类型

Tortoise ORM提供了丰富的字段类型，字段类型用于映射Python数据类型到数据库列。

from tortoise import fields

IntField

fields.IntField()

32位整数类型。

BigIntField

fields.BigIntField()

64位整数类型。

SmallIntField

fields.SmallIntField()

16位小整数类型。

FloatField

fields.FloatField()

浮点数类型。

DecimalField

fields.DecimalField(max_digits=整数最长位数, decimal_places=小数点精度)

高精度十进制数，必须指定max_digits、decimal_places参数。

CharField

fields.CharField(max_length=最大长度)

固定长度字符串类型，必须指定max_length参数。

TextField

fields.TextField()

长文本内容类型。

UUIDField

fields.UUIDField()

UUID(通用唯一标识符)类型。

可通过uuid库的uuid4生成随机UUID：default=uuid.uuid4。

DatetimeField

fields.DatetimeField()

日期时间类型。

常用参数（下面两个参数不能同时使用）：

auto_now_add=True：创建时自动设置当前时间。

auto_now=True：每次更新时自动设置当前时间。

DateField

fields.DateField()

仅日期(不含时间)类型。

TimeField

fields.TimeField()

仅时间(不含日期)类型。

BooleanField

fields.BooleanField()

布尔值类型，数据库中对应 BOOLEAN 或 TINYINT(1)。

ForeignKeyField

fields.ForeignKeyField("指向的模型路径")

多对一外键关系类型。

可选参数：related_name, on_delete

例如：fields.ForeignKeyField("models.Category", on_delete=fields.CASCADE)

OneToOneField

fields.OneToOneField("指向的模型路径")

一对一外键关系类型。

可选参数：related_name, on_delete

例如：fields.OneToOneField("models.Profile", on_delete=fields.CASCADE)

ManyToManyField

fields.ManyToManyField("指向的模型路径")

多对多外键关系类型。

可选参数：related_name, on_delete

例如：fields.ManyToManyField("models.Tag", related_name="posts")

JSONField

fields.JSONField()

JSON数据类型，PostgreSQL原生JSON，其他数据库存储为TEXT。

IntEnumField

fields.IntEnumField(enum_type=自定义枚举类名) 整数枚举类型。

from enum import IntEnum
from tortoise.models import Model
from tortoise import fields

class StatusEnum(IntEnum):
    DRAFT = 1
    PUBLISHED = 2
    ARCHIVED = 3

    # 自定义枚举显示名称，可通过"UserRoleEnum.DRAFT.label"查看对应自定义名称
    @property
    def label(self):
        labels = {
            "1": "草稿",
            "2": "已发布",
            "3": "已归档"
        }
        return labels[str(self.value)]

class User(Model):
    # 使用IntEnumField实现整数枚举
    status = fields.IntEnumField(
        enum_type=StatusEnum,
        default=StatusEnum.DRAFT,
        description="文章状态：1-草稿，2-已发布，3-已归档"
    )

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

CharEnumField

fields.CharEnumField(enum_type=自定义枚举类名) 整数枚举类型。

from enum import Enum

class UserRoleEnum(str, Enum):
    ADMIN = "admin"
    COMMON = "common"
    VIP = "vip"

    @property
    def label(self):
        labels = {
            "admin": "管理员用户",
            "common": "普通用户", 
            "vip": "VIP用户"
        }
        return labels[self.value]
    
class User(Model):
    # 使用CharEnumField实现字符串枚举
    role = fields.CharEnumField(
        enum_type=UserRoleEnum,
        max_length=20,
        default=UserRoleEnum.USER,
        description="用户角色"
    )

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

通用字段参数

所有字段都支持以下通用参数：

参数	说明	默认值
`null`	是否允许NULL值	False
`default`	默认值	None
`unique`	是否唯一	False
`index`	是否创建索引	False
`description`	字段描述（文档用）	None
`source_field`	自定义数据库列名	None

另外IntField和UUIDField还可以使用pk=True参数将字段定义为主键。

# Tortoise安装

# 安装Tortoise ORM，默认支持SQLite
pip install tortoise-orm

# 如果使用MySQL，需要安装MySQL异步模块
pip install aiomysql

# 如果使用PostgreSQL，需要安装PostgreSQL异步模块
pip install asyncpg

1
2
3
4
5
6
7
8

# 定义数据库模型

models.py 使用Model类来定义数据库表结构。

from tortoise.models import Model
from tortoise import fields

class User(Model):
    # Int字段类型，pk代表主键
    id = fields.IntField(pk=True)
    # 字符串字段类型，unique代表唯一键
    username = fields.CharField(max_length=50, unique=True)
    # max_length代表最大长度
    email = fields.CharField(max_length=100)
    # 日期时间字段类型，auto_now_add代表自动填充当前时间
    created_at = fields.DatetimeField(auto_now_add=True)
    # 布尔字段类型，default代表默认值
    is_active = fields.BooleanField(default=True)
    
    # 定义表模型的元数据
    class Meta:
        # 自定义表名
        table = "user"

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

# 初始化数据库连接

初始化数据连接后，可以直接通过ORM类进行增删改查操作。

from tortoise import Tortoise

TORTOISE_CONFIG = {
    # 数据库连接URL
    "connections": {"default": "sqlite://db.sqlite3"},
    "apps": {
        "models": {
            # 模型查找路径
            # 键是逻辑应用名可以任意，值是模块路径列表，建议用包，即模块放在models包内，会自动包含顶层的模型类，需要在包的__init__.py中导入子模块，如from .user import *
            "models": ["models"],
            "default_connection": "default",
        }
    }
}

async def init_db():
    # 初始化数据库连接
    await Tortoise.init(config=TORTOISE_CONFIG)

async def close_db():
    # 断开数据库连接，释放资源
    await Tortoise.close_connections()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

数据库URL格式：

SQLite: sqlite://db.sqlite3

PostgreSQL: postgres://user:password@localhost:5432/dbname

MySQL: mysql://user:password@localhost:3306/dbname

# 数据库迁移

TortoiseORM自带了数据库迁移工具aerich，用于管理数据库模式变更。

# 安装 aerich
pip install aerich

# 在项目根目录创建aerich.ini文件
[aerich]
# 个人用: app.core.settings:settings.tortoise_config
tortoise_orm = "db:TORTOISE_ORM"
migrations_dir = "migrations"

# 使用aerich
# 初始化迁移，指定db.py内的TORTOISE_CONFIG配置信息
# 个人用: app.core.settings.settings.tortoise_config
aerich init -t db.TORTOISE_CONFIG
aerich init-db
# 生成迁移文件，迁移名称风格可以根据按自己喜好决定
aerich migrate --name 000001
# 应用迁移
aerich upgrade
# 回滚迁移
aerich downgrade

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

# 查询数据

# 模型查询

[ORM模型类名].filter(字段名=查询值...)

按条件取出所有匹配到的行的数据对象，返回 [<数据对象>, <数据对象>...]

PS：另外可以使用pk参数来指代当前表的主键字段。

[ORM模型类名].all()

取出表所有数据。

# 查询所有活跃用户
active_users = await User.filter(is_active=True).all()

# 按条件查询，未找到返回None
user = await User.get_or_none(username="alice")

# 复杂查询（AND/OR）
from tortoise.expressions import Q
recent_users = await User.filter(
    Q(is_active=True) & Q(created_at__gte="2023-01-01")
).order_by("-created_at").limit(10)

# 仅获取特定字段
user_names = await User.all().values("username", "email")

1
2
3
4
5
6
7
8
9
10
11
12
13
14

另外你还可以先根据条件去构建QuerySet对象(查询计划)，最后再通过await去数据库中查询，只要不await就不会获取数据。

# 查询方法

raw("""SQL语句""")

通过原生SQL语句查询，返回QuerySet对象，列表套字典。

all()

查询表所有数据，返回QuerySet对象，列表套字典。

filter(字段=值,...)

带有过滤条件的查询数据，返回QuerySet对象，列表套字典。

filter内的多个参数默认是AND关系，需要借助于Q查询才可以进行其他关系的过滤(OR、NOT等)。

get(字段=值,...)

带有过滤条件的查询数据，直接拿数据对象。

数据不存在时会直接报错，不推荐使用。

get_or_none(字段=值,...)

带有过滤条件的查询数据，直接拿数据对象。

数据不存在时不会报错，而是返回None，推荐使用。

first()

取出QuerySet里面的第一个元素。

last()

取出QuerySet里面的最后一个元素。

values("字段名1", "字段名2",...)

只取出指定的字段。

可以传多个值，指定多个要取出的字段，返回QuerySet对象，列表套字典。

values_list("字段名1", "字段名2",...)

只取出指定的字段。

可以传多个值，指定多个要取出的字段，返回QuerySet对象，列表套元组。

如果只查询单个字段，可以指定 flat=True 将值直接放到一个列表中返回。

distinct()

结果去重，需要配合values过滤掉主键后使用。

例如：models.Users.objects.values("username").distinct()

order_by()

结果排序，会以指定的字段进行排序，默认是升序。

如果指定字段名参数开头加上"-"号，就是降序，例如：order_by("-username")

reverse()

将数据反转，前提是数据是排过序的，否则反转会无效。

例如：models.Users.objects.all().order_by("username").reverse()

count()

对当前数据的个数进行统计。

exclude()

排除指定条件的数据行。

exists()

判断查询的数据是否存在，返回布尔值。基本用不到，因为数据本身就能当布尔值。

# 字段查找操作符

filter() 方法支持丰富的字段查找操作符，语法采用双下划线 __ 连接字段名和操作符。

Model.filter(字段名__操作符=值)

另外get()、get_or_none()方法则不支持使用，可以用filter().first()代替。

操作符	含义	SQL 等效	示例
`exact`	精确匹配（默认）	`=`	`name="Alice"`
`iexact`	精确匹配（不区分大小写）	`LOWER(col)=LOWER('alice')`	`name__iexact="alice"`
`contains`	包含子串	`LIKE '%abc%'`	`title__contains="bug"`
`icontains`	包含子串（不区分大小写）	`ILIKE '%bug%'`	`title__icontains="BUG"`
`in`	在列表中	`IN (1, 2, 3)`	`id__in=[1, 2, 3]`
`not_in`	不在列表中	`NOT IN (1, 2)`	`status__not_in=["deleted"]`
`gt`	大于	`>`	`age__gt=18`
`gte`	大于等于	`>=`	`price__gte=100`
`lt`	小于	`<`	`score__lt=60`
`lte`	小于等于	`<=`	`count__lte=10`
`startswith`	以...开头	`LIKE 'abc%'`	`email__startswith="admin"`
`istartswith`	以...开头（不区分大小写）	`ILIKE 'admin%'`	`name__istartswith="ALICE"`
`endswith`	以...结尾	`LIKE '%.com'`	`domain__endswith=".com"`
`iendswith`	以...结尾（不区分大小写）	`ILIKE '%.COM'`	`email__iendswith=".COM"`
`isnull`	是否为 NULL	`IS NULL` / `IS NOT NULL`	`deleted_at__isnull=True`
`regex`	正则匹配（区分大小写）	`~ 'pattern'` (PG)	`text__regex=r'^\d{3}-\d{4}$'`
`iregex`	正则匹配（不区分大小写）	`~* 'pattern'` (PG)	`name__iregex=r'^(alice)$`

# 聚合查询

在Tortoise ORM中，聚合查询（如 COUNT、SUM、AVG、MAX、MIN）是通过 .annotate() + 聚合函数 实现的。

不过Tortoise的聚合功能相对基础，多表聚合等复杂场景建议使用原生SQL。

# 聚合函数定义在tortoise.aggregation中
from tortoise.aggregation import Count, Sum, Avg, Max, Min

# 基本形式，如果不加group_by()则是全表聚合只返回一行
Model.annotate(别名=聚合函数("字段")).group_by("分组字段")

# 例如: 统计总用户数
result = await User.annotate(count=Count("id")).first()
print(result.count)

1
2
3
4
5
6
7
8
9

# F表达式

F表达式可以允许在数据查询或数据更新时引用数据库中的字段值，而不是Python变量。

注意：TortoiseORM中的 F 不能用于 .filter() 中的字段比较，例如 age__gt=F("age") 时不允许的，所以 F 基本只在更新操作时使用。

# 导入
from tortoise.expressions import F

# 基本形式
F("字段名")

# 例如:
# 将数据库中views的值自增1
await Article.filter(id=1).update(views=F("views") + 1)

1
2
3
4
5
6
7
8
9

# Q对象

Q对象可以用于构建复杂的逻辑查询条件，支持：AND(默认)、|(或)、~(非)。

# 导入
from tortoise.queryset import Q

# 基本形式
Q(字段名=值)
Q(字段名__条件=值)

# 例如:
# OR查询
users = await User.filter(Q(name="Alice") | Q(name="Bob"))

# Q对象的条件可以混合使用
users = await User.filter(is_active=True, ~Q(role="admin"))

# Q对象中可以使用字段查找操作符
users = await User.filter(Q(age__gte=18) | Q(name="Bob"))

# Q对象还可以嵌套
posts = await Post.filter(
    Q(status="published") & (Q(author_id=1) | Q(reviewer_id=1))
)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

# 惰性查询特性

ORM查询只有在引用获取查询结果时才会执行语句。如果只是书写了ORM查询语句，而没有用到该语句所查询出来的结果，那么ORM会自动识别，根本不执行该语句。

# 例如：

# 书写ORM查询语句，不执行查询
res = models.Users.objects.all()
# 引用查询结果，执行查询
print(res)

# 每获取一次对象的数据，就会执行一次查询语句
res = models.Users.objects.all()
for i in res:
    print(i.name)

1
2
3
4
5
6
7
8
9
10
11

# 创建数据

[ORM模型类名].create(字段名=值...)

返回创建的行的数据对象。

# 创建单条记录
user = await User.create(
    username="alice",
    email="alice@example.com"
)

# 批量创建
await User.bulk_create([
    User(username="bob", email="bob@example.com"),
    User(username="charlie", email="charlie@example.com")
])

1
2
3
4
5
6
7
8
9
10
11

# 更新数据

[ORM模型类名].filter(字段=值).update(字段=值...)

# 更新单条记录
await User.filter(id=1).update(is_active=False)

# 直接对单一对象修改
user = await User.filter(id=1).first()
data.is_active = False
# 需要保存修改
data.save()

# 使用F表达式
from tortoise.functions import F
await User.filter(id=1).update(balance=F('balance') + 100)

1
2
3
4
5
6
7
8
9
10
11
12

# 删除数据

res = [ORM模型类名].filter(字段名=查询值...).delete()

会返回一个元组，其中有删除了多少行的信息。

# 先通过filter查询出所有匹配的记录，然后调用delete()删除
await User.filter(id=1).delete()

# 直接对单一对象删除
user = await User.filter(id=1).first()
data.delete()

1
2
3
4
5
6

# 多表关系

表关系分为主表与从表，同一数据库中B表的外键与A表的主键相对应，则在A与B表的关系中，A表为主表，B表为从表。

Tortoise ORM和Django ORM一样，也支持三种关系类型字段，即一对一关系、一对多关系、多对多关系。

指定的ORM模型路径需要是Tortoise配置中设置的APP名，一般为models，加目标的模型名。

另外ORM在数据库中创建关系字段时会自动加上_id后缀，无需在在模型中添加。

# 一对一关系

一对一关系中，外键字段在任意一方均可，但推荐建在"从属实体"上。

例如一个用户对应一个身份证，用户是主要实体，外键要建在身份证表上。

字段名 = models.OneToOneField("ORM模型路径", on_delete=models.CASCADE)

书写ORM模型路径即可，会自动关联到目标模型的主键字段。

class Profile(Model):
    id = fields.IntField(pk=True)
    user = fields.OneToOneField("models.User", related_name="profile",  on_delete=fields.CASCADE)

1
2
3

# 一对多关系

一对多关系中，一是主表，多是从表。外键需要建在多的从表上。

例如一个用户可以写多篇文章，那么外键要建在文章表上。

字段名 = models.ForeignKeyField("ORM模型路径", on_delete=models.CASCADE)

class BlogPost(Model):
    id = fields.IntField(pk=True)
    title = fields.CharField(max_length=200)
    author = fields.ForeignKeyField("models.User", related_name="posts")

1
2
3
4

# 多对多关系

多对多关系需要通过一个中间表实现，中间表内包含 两个表的外键 和 可选元数据(创建时间等)。

中间表的命名规范是用两个表名加下划线_连接，可以按字母顺序排序，也可以按"主体-附属"顺序排序，例如：user_role。

# 创建中间表
class UserRole(Model):
    user = fields.ForeignKeyField("models.User", related_name="roles")
    role = fields.ForeignKeyField("models.Role", related_name="users")
    create_time = fields.DatetimeField(auto_now_add=True)

    # 防止重复
    class Meta:
        table = "user_role"
        unique_together = ("user", "role")

# 查询时推荐使用两步查询法
# 先通过user_id从中间表查询出匹配的所有role_id列表
role_ids = await UserRole.filter(user=user.id).values_list("role_id", flat=True)
# 去role表中通过in过滤查询出结果
roles = await Role.filter(id__in=role_ids)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

# on_delete

on_delete参数用于定义关联的主表记录被删除时，数据库应如何处理从表中的相关记录。

它有以下几种选项：

选项	行为说明
`fields.CASCADE`	级联删除(最常用)：删除主表记录时，自动删除所有关联的从表记录
`fields.RESTRICT`	禁止删除：如果存在关联记录，禁止删除主表记录（抛出 DB 异常）
`fields.SET_NULL`	设为 NULL：删除主表记录时，将外键字段设为 `NULL`（要求字段允许 `null=True`）
`fields.SET_DEFAULT`	设为默认值：将外键设为默认值（需定义 `default`，且数据库支持）
`fields.NO_ACTION`	无动作：不采取任何措施（依赖数据库默认行为，通常等同于 `RESTRICT`）

related_name 参数用于定义反向关系的名称。主要用于一对一和一对多关系中，能使被关联的模型可以通过名称反向访问关联对象。

会在被关联模型中，创建一个叫指定名称的属性，用来访问这个属性可以访问关联的当前模型数据。

如果不指定，则会自动生成一个反向关系名，格式为 [当前模型名]_set，建议自己书写更加直观。

class User(Model):
    name = fields.CharField(50)

class Post(Model):
    title = fields.CharField(100)
    user = fields.ForeignKeyField("models.User", related_name="posts")

# 先查询用户对象
user = await User.get_or_none(name="Alice")
# 直接从user模型查询所有Alice的posts数据
posts = await user.posts.all()

1
2
3
4
5
6
7
8
9
10
11

# 子查询

可以通过数据行对象的外键字段查询外表的数据。

[表数据行对象] = [表ORM类名].get_or_none(过滤条件)

res = [表数据行对象].[表的外键字段名].[外键表的字段名]

book = await Book.get_or_none(pk=1)
# 会用该行的author_id去author表查询name字段返回
res = book.author.name

1
2
3

该方法是用于高效预加载关联对象的方法，主要解决在循环中访问关联字段时可能出现的N+1查询问题。

N+1查询问题指在获取主表的N条记录后，每访问一次记录的未预加载关联字段，ORM就会发起一次数据库查询，总共导致N+1次查询。

# 不使用prefetch_related的情况

# 第1次查询，获取所有文章
posts = await Post.all()

# 然后循环访问每个文章的作者
for post in posts:
    # 每次都触发1次数据库查询
    print(post.author.name)

# ========================

# 使用prefetch_related的情况
# 总共只会查询两次:
#   第1次: 获取所有文章
#   第2次: 批量获取所有用到的作者并缓存
posts = await Post.all().prefetch_related("author")

for post in posts:
    # 直接从缓存中获取数据，无额外查询
    print(post.author.name)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

# 事务处理

from tortoise.transactions import in_transaction

async def transfer_funds(sender_id, receiver_id, amount):
    async with in_transaction() as conn:
        sender = await User.get(id=sender_id)
        receiver = await User.get(id=receiver_id)
        
        await sender.update(balance=sender.balance - amount)
        await receiver.update(balance=receiver.balance + amount)
        
        # 事务自动提交，如果出错会自动回滚

1
2
3
4
5
6
7
8
9
10
11

#Python #模块 #Tortoise #异步 #ORM框架

← FastAPI中间件与依赖 FastAPI实现用户管理→

Tortoise ORM

# Tortoise ORM

# 介绍

# 特点

# 字段类型

# Tortoise安装

# 定义数据库模型

# 初始化数据库连接

# 数据库迁移

# 查询数据

# 模型查询

# 查询方法

# 字段查找操作符

# 聚合查询

# F表达式

# Q对象

# 惰性查询特性

# 创建数据

# 更新数据

# 删除数据

# 多表关系

# 一对一关系

# 一对多关系

# 多对多关系

# on_delete

# related_name

# 子查询

# prefetch_related

# 事务处理