Tortoise ORM 完整使用笔记（从配置到实战）

一、概述

Tortoise ORM 是 Python 异步 ORM 框架，对标 Django ORM，支持 SQLite、MySQL、PostgreSQL 等主流数据库，核心特点：

  • 纯异步设计，适配 FastAPI/Starlette 等异步框架；
  • 语法贴近 Django ORM，学习成本低；
  • 支持模型关联、事务、索引、原生 SQL 等核心功能；
  • 内置数据验证、迁移工具（tortoise-orm[asyncmy] + aerich）。

二、环境准备

1. 安装依赖

# 基础安装（适配不同数据库需加对应驱动）
pip install tortoise-orm

# 适配 MySQL（asyncmy 驱动）
pip install tortoise-orm[asyncmy]

# 适配 PostgreSQL（asyncpg 驱动）
pip install tortoise-orm[asyncpg]

# 迁移工具（数据库表结构同步）
pip install aerich

2. 支持的数据库

数据库 驱动依赖 配置标识
SQLite 无需额外驱动 sqlite://
MySQL/MariaDB asyncmy mysql://
PostgreSQL asyncpg postgres://

三、核心配置（初始化）

1. 基础配置格式

Tortoise ORM 需先初始化数据库连接，支持同步初始化和异步初始化（推荐异步）。

（1）异步初始化（推荐，适配 FastAPI 等）
from tortoise import Tortoise, run_async

# 异步初始化函数
async def init_db():
    await Tortoise.init(
        db_url="mysql://用户名:密码@IP:端口/数据库名?charset=utf8mb4",  # 数据库连接地址
        modules={"models": ["app.models"]}  # 指定模型所在模块（多个模块用列表）
    )
    # 生成数据库表（开发环境用，生产环境建议用 aerich 迁移）
    await Tortoise.generate_schemas()

# 执行初始化（单独测试用，FastAPI 中可在 startup 事件中调用）
if __name__ == "__main__":
    run_async(init_db())
（2）不同数据库的 db_url 示例
# SQLite（文件型，无需服务）
db_url = "sqlite://./test.db"  # 相对路径
db_url = "sqlite:///绝对路径/test.db"  # 绝对路径

# MySQL（asyncmy 驱动）
db_url = "mysql://root:123456@127.0.0.1:3306/tortoise_demo?charset=utf8mb4"

# PostgreSQL（asyncpg 驱动）
db_url = "postgres://postgres:123456@127.0.0.1:5432/tortoise_demo"
（3）FastAPI 中集成（生命周期管理）
from fastapi import FastAPI
from tortoise.contrib.fastapi import register_tortoise

app = FastAPI()

# 注册 Tortoise ORM（自动处理初始化/关闭）
register_tortoise(
    app,
    db_url="mysql://root:123456@127.0.0.1:3306/tortoise_demo",
    modules={"models": ["app.models"]},
    generate_schemas=True,  # 开发环境自动生成表，生产环境关闭
    add_exception_handlers=True,  # 自动添加 ORM 异常处理器
)

2. 高级配置（可选）

await Tortoise.init(
    db_url="mysql://root:123456@127.0.0.1:3306/tortoise_demo",
    modules={"models": ["app.models"]},
    # 额外配置
    config={
        "connections": {
            "default": {
                "engine": "tortoise.backends.mysql",
                "credentials": {
                    "host": "127.0.0.1",
                    "port": 3306,
                    "user": "root",
                    "password": "123456",
                    "database": "tortoise_demo",
                    "charset": "utf8mb4",
                    "minsize": 1,  # 连接池最小连接数
                    "maxsize": 10,  # 连接池最大连接数
                }
            }
        },
        "apps": {
            "models": {
                "models": ["app.models"],
                "default_connection": "default",
            }
        }
    }
)

四、模型定义（核心）

1. 基础模型结构

所有模型需继承 tortoise.Model，字段通过 tortoise.fields 定义。

from tortoise import Model, fields
from datetime import datetime

class User(Model):
    # 核心字段（常用类型+参数）
    id = fields.IntField(pk=True)  # 主键，默认自增
    username = fields.CharField(max_length=50, unique=True, comment="用户名")  # 唯一字符串
    email = fields.CharField(max_length=100, null=True, default=None, comment="邮箱")  # 允许为空
    age = fields.IntField(min_value=0, max_value=120, default=0, comment="年龄")  # 数值范围限制
    is_active = fields.BooleanField(default=True, comment="是否激活")
    create_time = fields.DatetimeField(auto_now_add=True, comment="创建时间")  # 仅创建时赋值
    update_time = fields.DatetimeField(auto_now=True, comment="更新时间")  # 每次保存时更新

    # 元数据配置（表名、索引、排序等）
    class Meta:
        table = "user"  # 数据库表名（默认是模型名小写+复数，如 user → users）
        table_description = "用户表"  # 表备注
        indexes = [("username", "email")]  # 联合索引
        unique_together = [("username", "email")]  # 联合唯一约束
        ordering = ["-create_time"]  # 默认排序（- 表示降序）

2. 常用字段类型

字段类型 作用 常用参数
IntField 整数 pk, default, min_value, max_value
CharField 字符串 max_length, unique, null, default
TextField 长文本（无长度限制） null, default
BooleanField 布尔值 default
DatetimeField 日期时间 auto_now, auto_now_add, null
DateField 日期 auto_now, auto_now_add, null
FloatField 浮点数 min_value, max_value, default
DecimalField 高精度小数 max_digits, decimal_places
ForeignKeyField 外键（一对多） to, related_name, on_delete
ManyToManyField 多对多 to, related_name

3. 字段核心参数

参数 作用
pk=True 标记为主键（IntField 主键默认自增）
unique=True 字段唯一约束
null=True 允许字段为 NULL（默认不允许）
default 字段默认值
comment 字段备注（生成表结构时同步到数据库）
auto_now 每次保存时自动更新为当前时间（仅 Datetime/Date）
auto_now_add 创建时自动赋值为当前时间（仅 Datetime/Date）

五、CRUD 操作（核心方法）

所有 ORM 操作均为异步，需加 await 执行。

1. 创建数据（Create）

方法 作用 示例
create(** kwargs) 新增单条记录 user = await User.create(username="test", email="test@xxx.com")
bulk_create(列表) 批量新增（高效） users = [User(username="u1"), User(username="u2")]<br>await User.bulk_create(users)
get_or_create() 查不到则创建，返回 (实例, 是否创建) user, created = await User.get_or_create(username="test", defaults={"email": "test@xxx.com"})
create_or_update() 按条件更新，不存在则创建 await User.create_or_update(username="test", defaults={"email": "new@xxx.com"})

2. 查询数据（Read）

（1）基础查询
方法 作用 示例
all() 查询所有记录，返回 QuerySet users = await User.all()
get(** kwargs) 查单条，找不到抛 DoesNotExist 异常 user = await User.get(username="test")
get_or_none(** kwargs) 查单条，找不到返回 None（推荐） user = await User.get_or_none(username="test", is_active=True)
filter(** kwargs) 按条件筛选，返回 QuerySet active_users = await User.filter(is_active=True)
exclude(** kwargs) 按条件排除，返回 QuerySet inactive_users = await User.exclude(is_active=True)
first() 取 QuerySet 第一条记录 first_user = await User.filter(is_active=True).first()
last() 取 QuerySet 最后一条记录 last_user = await User.filter(is_active=True).last()
count() 统计记录数 count = await User.filter(is_active=True).count()
exists() 判断 QuerySet 是否有记录（高效） has_test = await User.filter(username="test").exists()
（2）高级筛选（条件表达式）
语法 作用 示例
field__contains 模糊匹配（包含） await User.filter(username__contains="test")
field__icontains 模糊匹配（忽略大小写） await User.filter(email__icontains="gmail.com")
field__startswith 以指定字符串开头 await User.filter(username__startswith="u_")
field__endswith 以指定字符串结尾 await User.filter(email__endswith="@qq.com")
field__in 字段值在列表中 await User.filter(id__in=[1,2,3])
field__gt/lt 大于/小于 await User.filter(age__gt=18) / await User.filter(age__lt=30)
field__gte/lte 大于等于/小于等于 await User.filter(age__gte=18) / await User.filter(age__lte=30)
field__isnull 字段是否为 NULL await User.filter(email__isnull=True)（查无邮箱的用户）
（3）排序 & 分页
方法 作用 示例
order_by("字段") 排序（- 表示降序） await User.filter(is_active=True).order_by("-create_time")
offset(n) 跳过前 n 条（分页） await User.all().offset(10)（跳过前 10 条）
limit(n) 取 n 条（分页） await User.all().limit(10)（取前 10 条）
offset+limit 分页组合 await User.all().offset(10).limit(10)（第 2 页，每页 10 条）
（4）聚合查询

需导入 tortoise.functions 中的聚合函数：

from tortoise.functions import Count, Sum, Max, Min, Avg

# 1. annotate：新增计算字段（每条记录）
users = await User.annotate(post_count=Count("posts")).all()  # 统计每个用户的帖子数
for user in users:
    print(user.username, user.post_count)

# 2. aggregate：全局聚合（返回字典）
result = await User.aggregate(
    total=Count("id"),
    max_age=Max("age"),
    avg_age=Avg("age")
)
print(result)  # {"total": 100, "max_age": 80, "avg_age": 28.5}
（5）原生 SQL 查询

复杂查询可直接执行原生 SQL：

# 方法1：raw()（返回模型实例）
users = await User.raw("SELECT * FROM user WHERE age > %s", [18])

# 方法2：execute_query()（返回原始数据）
from tortoise.connection import connections
conn = connections.get("default")
rows = await conn.execute_query("SELECT username FROM user WHERE is_active = 1")
print(rows)  # {"rows": [("test1",), ("test2",)], "columns": ["username"]}

3. 更新数据（Update）

方法 作用 示例
save() 实例更新（先查后改） user = await User.get(id=1)<br>user.age = 20<br>await user.save()
update(** kwargs) QuerySet 批量更新（高效） await User.filter(username="test").update(email="new@xxx.com")
bulk_update(列表, 字段列表) 批量更新指定字段 user1 = await User.get(id=1)<br>user2 = await User.get(id=2)<br>user1.age=21<br>user2.age=22<br>await User.bulk_update([user1, user2], ["age"])

4. 删除数据（Delete）

方法 作用 示例
delete() 实例/QuerySet 删除（物理删除） # 单条删除<br>user = await User.get(id=1)<br>await user.delete()<br># 批量删除<br>await User.filter(is_active=False).delete()
bulk_delete() 批量删除（高效） await User.filter(id__in=[1,2]).bulk_delete()
逻辑删除（推荐） 更新 is_deleted 字段 await User.filter(id=1).update(is_deleted=True)

六、关联关系操作

1. 一对多（ForeignKey）

（1）定义模型
# 帖子模型（关联 User）
class Post(Model):
    id = fields.IntField(pk=True)
    title = fields.CharField(max_length=100)
    content = fields.TextField()
    author = fields.ForeignKeyField("models.User", related_name="posts", on_delete=fields.CASCADE)  # 关联 User
    create_time = fields.DatetimeField(auto_now_add=True)

    class Meta:
        table = "post"
  • related_name="posts"：反向查询别名（User 实例可通过 user.posts 查所有帖子）；
  • on_delete=fields.CASCADE：级联删除（删除用户时，关联帖子也删除），可选值：
    • CASCADE：级联删除；
    • SET_NULL：设为 NULL（需字段允许 null）；
    • SET_DEFAULT：设为默认值；
    • RESTRICT：禁止删除（有关联数据时抛异常）。
（2）关联查询
# 正向查询（Post → User）
post = await Post.get(id=1)
await post.fetch_related("author")  # 预加载作者（解决 N+1 查询）
print(post.author.username)

# 反向查询（User → Post）
user = await User.get(id=1)
posts = await user.posts.filter(title__contains="test").all()  # 查用户的所有含 test 的帖子

# 预加载关联数据（高效）
users = await User.all().fetch_related("posts")  # 一次性查所有用户 + 关联帖子

2. 多对多（ManyToMany）

（1）定义模型
# 标签模型
class Tag(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=20, unique=True)

    class Meta:
        table = "tag"

# 帖子-标签 多对多（Tortoise 自动生成中间表）
class Post(Model):
    id = fields.IntField(pk=True)
    title = fields.CharField(max_length=100)
    tags = fields.ManyToManyField("models.Tag", related_name="posts")  # 多对多关联 Tag

    class Meta:
        table = "post"
（2）多对多操作
# 1. 添加标签
post = await Post.get(id=1)
tag1 = await Tag.get(name="技术")
tag2 = await Tag.get(name="Python")
await post.tags.add(tag1, tag2)  # 给帖子添加多个标签

# 2. 移除标签
await post.tags.remove(tag1)

# 3. 查询关联标签
tags = await post.tags.all()
print([tag.name for tag in tags])

# 4. 反向查询（标签 → 帖子）
tag = await Tag.get(name="Python")
posts = await tag.posts.filter(title__contains="ORM").all()

3. 一对一（OneToOne）

类似一对多，用 OneToOneField 定义，查询逻辑与一对多一致。

七、事务操作

确保多个操作原子性（要么都成功，要么都失败）。

1. 上下文管理器（推荐）

from tortoise.transactions import in_transaction

async def create_user_and_post():
    async with in_transaction():
        # 事务内的操作
        user = await User.create(username="test")
        await Post.create(title="test post", author=user)
    # 事务结束自动提交，异常则回滚

2. 装饰器方式

from tortoise.transactions import atomic

@atomic
async def create_user_and_post():
    user = await User.create(username="test")
    await Post.create(title="test post", author=user)

八、常用工具方法（速查）

方法 作用 示例
values(*fields) 返回字典列表（仅含指定字段） user_dicts = await User.all().values("id", "username")
values_list(*fields, flat=False) 返回元组列表 user_ids = await User.all().values_list("id", flat=True) # 一维列表
distinct(*fields) 去重查询 unique_emails = await User.all().distinct("email").values_list("email")
only(*fields) 仅加载指定字段（减少数据传输） users = await User.all().only("username", "email")
exclude(*fields) 排除指定字段 users = await User.all().exclude("create_time", "update_time")
select_related() 预加载外键关联（一对一/一对多，比 fetch_related 高效） posts = await Post.all().select_related("author")

九、数据库迁移（aerich）

开发中模型变更后，需同步到数据库（替代 generate_schemas()）：

# 初始化迁移配置
aerich init -t app.config.TORTOISE_ORM  # TORTOISE_ORM 是你的配置字典

# 初始化迁移目录
aerich init-db

# 模型变更后生成迁移文件
aerich migrate --name update_user_model

# 执行迁移（同步到数据库）
aerich upgrade

# 回滚迁移
aerich downgrade

十、常见问题 & 避坑点

  1. 异步操作必须加 await：所有 ORM 方法（all()/create()/filter() 等）返回 QuerySet，需加 await 才会执行；
  2. QuerySet 懒加载：User.filter(is_active=True) 只是构建查询语句，加 await 才会真正查数据库；
  3. N+1 查询问题：关联查询时必须用 fetch_related()/select_related() 预加载，避免循环查询；
  4. get() 抛异常：优先用 get_or_none()，避免 DoesNotExist 异常；
  5. 批量操作优先用 bulk_create/bulk_update：比循环 create/save 高效 10 倍以上；
  6. 事务内操作不立即生效：事务未提交前，外部查不到未提交的数据。

十一、总结

  1. 核心流程：配置初始化 → 定义模型 → 异步 CRUD → 关联查询/事务；
  2. 高频方法：get_or_none()/filter()/create()/update()/fetch_related() 占日常使用 80%；
  3. 性能优化：批量操作、预加载关联、仅加载必要字段；
  4. 生产规范：用 aerich 做迁移，避免直接用 generate_schemas()；逻辑删除替代物理删除。
# python
Logo
2048 AI社区

有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

更多推荐

LangChain+Gradio+千问大模型实现AI聊天助手

1. 非流式输出模式。

avatar 2048 AI社区
cover

Stripe花75亿美元买了个“中间商“，OpenRouter被支付巨头收编了

avatar 2048 AI社区

MoneyPrinterTurbo：让中小企业把短视频生产成本压到接近零 ｜SSP Github Daily

MoneyPrinterTurbo 是务实的“API 编排”项目——不自研模型、不赌未来技术，只用今天成熟的 API，把短视频生产成本压到接近零，让中小企业从“单条手工剪辑”转向“批量内容实验”。

avatar 2048 AI社区