Tortoise-ORM 快速入门：像写 django 一样写异步数据库操作

📂 所属阶段：第三阶段 — 数据持久化（数据库篇）
🔗 相关章节：SQLAlchemy 2.0 实战 · Redis 集成 · FastAPI 路由

1. 为什么选 Tortoise-ORM？和 SQLAlchemy 说再见

如果把 SQLAlchemy 比作一把功能完备的瑞士军刀，那么 Tortoise-ORM 就是为 Python 异步开发量身打造的折叠剪刀——轻巧、直观、开箱即用。对于中小型项目和快速原型，它几乎可以零学习成本地让你从 django ORM 无缝过渡到异步世界。

1.1 一分钟对比 SQLAlchemy 2.0

核心痛点Tortoise-ORM 的解法SQLAlchemy 2.0 的解法
学习成本完全复用 django ORM 的 API 风格，没有额外的心智负担2.0 版同时提供 Core 和 ORM 两套范式，概念层次多，入门常需 1–2 周
异步体验从设计之初就是原生 async/await，不存在同步到异步的切换割裂2.0 才真正支持原生异步，但底层仍基于同步引擎，必须严格管理异步会话生命周期
配置复杂度一个 Python 字典就能定义连接、模型、时区需要手动构建异步引擎、会话工厂，并处理上下文传递

简单说：如果你喜欢 django ORM 的表达方式，又需要原生异步支持，Tortoise-ORM 就是最好的答案。

1.2 一行命令搞定开发环境

根据你使用的数据库选择依赖，不要全部安装：

# 开发/测试首选：SQLite（无需额外服务）
pip install tortoise-orm[sqlite]

# 生产环境推荐：PostgreSQL（性能最优）
pip install tortoise-orm[asyncpg] asyncpg

2. 初始化：两个文件让项目跑起来

Tortoise-ORM 的初始化非常清爽：分离配置与启动逻辑，整个流程只需要两个文件。

2.1 配置文件 tortoise_config.py

将连接信息、模型路径、时区等集中管理，后续只需导入一个字典即可：

from tortoise import Tortoise

TORTOISE_ORM = {
    "connections": {
        "default": {
            "engine": "tortoise.backends.sqlite",       # 开发用 SQLite
            # "engine": "tortoise.backends.asyncpg",   # 生产用 PostgreSQL
            "credentials": {
                "file_path": "./dev.db",                # SQLite 文件路径
                # "host": "localhost",
                # "port": 5432,
                # "user": "dev_user",
                # "password": "dev_pwd",
                # "database": "tortoise_demo",
            }
        }
    },
    "apps": {
        # 自定义 app 名称，后续模型引用时使用 `app名.模型类名`
        "demo": {
            "models": ["__main__", "demo_models"],      # 模型所在的模块路径
            "default_connection": "default",
        }
    },
    "use_tz": False,          # 国内项目一般无需 UTC 转换
    "timezone": "Asia/Shanghai",
}

2.2 在 FastAPI 中一键挂载

FastAPI 社区提供了 register_tortoise 工具，自动处理数据库连接的启动、关闭以及建表操作：

from fastapi import FastAPI
from tortoise.contrib.fastapi import register_tortoise
from tortoise_config import TORTOISE_ORM

app = FastAPI(title="Tortoise + FastAPI CRUD 演示")

register_tortoise(
    app,
    config=TORTOISE_ORM,
    generate_schemas=True,             # 开发时自动建表（生产环境建议使用 Aerich 迁移工具）
    add_exception_handlers=True,       # 自动捕获模型验证、不存在等异常
)

提示：生产环境中应使用迁移工具 Aerich 来管理表结构变更，而不是依赖 generate_schemas=True。

3. 模型定义：熟悉的 django ORM 配方

新建 demo_models.py，定义一套经典的博客模型：作者、标签、文章，涵盖外键、多对多、自动时间戳等常用字段。

from tortoise import fields
from tortoise.models import Model

class Author(Model):
    id = fields.IntField(pk=True, autoincrement=True)
    name = fields.CharField(max_length=50, description="作者姓名")
    email = fields.CharField(max_length=100, unique=True, description="唯一邮箱")
    # 反向关联提示（非必需，但能提供 IDE 自动补全）
    articles: fields.ReverseRelation["Article"]

    class Meta:
        table = "authors"
        ordering = ["name"]

    def __str__(self):
        return self.name


class Tag(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=30, unique=True, description="标签名")
    articles: fields.ReverseRelation["Article"]

    def __str__(self):
        return self.name


class Article(Model):
    id = fields.IntField(pk=True)
    title = fields.CharField(max_length=200, index=True, description="文章标题（加索引加速搜索）")
    content = fields.TextField(description="文章正文")
    views = fields.IntField(default=0, description="阅读量")
    is_published = fields.BooleanField(default=False, description="是否发布")
    created_at = fields.DatetimeField(auto_now_add=True, description="创建时间")
    updated_at = fields.DatetimeField(auto_now=True, description="更新时间")

    # 外键：删除作者时级联删除其所有文章
    author = fields.ForeignKeyField(
        "demo.Author",            # 格式：app名.模型类名
        related_name="articles",
        on_delete=fields.CASCADE,
    )
    # 多对多：通过 article_tag 中间表关联标签
    tags = fields.ManyToManyField(
        "demo.Tag",
        related_name="articles",
        through="article_tag",
    )

    class Meta:
        table = "articles"
        ordering = ["-created_at"]   # 最新文章排在最前

定义完成后，Tortoise-ORM 会根据模型自动生成对应的数据库表，外键和多对多关系也会一并建立，完全不需要手写 SQL。

4. CRUD 实战：记住一个 await 就够了

所有数据库操作都必须在异步函数中执行，并在调用前加上 await。下面的例子使用纯 Tortoise-ORM 接口，不依赖任何 Web 框架，方便你在任意地方复用。

4.1 创建（Create）

# 单条创建
new_author = await Author.create(
    name="张三",
    email="zhangsan@example.com"
)

# 批量创建（比循环单个创建快 10 倍以上）
await Tag.bulk_create([
    Tag(name="Python"),
    Tag(name="FastAPI"),
    Tag(name="Tortoise-ORM"),
])

# 创建文章并关联作者和标签
new_article = await Article.create(
    title="Tortoise-ORM 入门第一弹",
    content="今天我们来学习 Tortoise-ORM ...",
    author=new_author   # 也可以直接传 author_id=1
)

python_tag = await Tag.get(name="Python")
tortoise_tag = await Tag.get(name="Tortoise-ORM")
await new_article.tags.add(python_tag, tortoise_tag)

4.2 读取（Read）

关键原则：如果要访问外键或多对多关联字段，务必使用 prefetch_related() 提前加载数据，否则容易引发 N+1 查询问题。

# 查询单条：若不存在则抛出 DoesNotExist 异常
author = await Author.get(id=1)

# 安全查询：找不到返回 None
article = await Article.get_or_none(id=999)

# 过滤查询
published_articles = await Article.filter(is_published=True)

# 跨表过滤（使用双下划线语法）
python_articles = await Article.filter(tags__name="Python").prefetch_related("author", "tags")

# 分页与排序
page1 = await Article.all().offset(0).limit(10).order_by("-views")

# 只查询需要的字段（减少数据传输）
titles = await Article.all().only("id", "title")

# 聚合统计：计算每个作者的文章数
from tortoise.functions import Count

author_stats = await Author.annotate(article_count=Count("articles"))
for stat in author_stats:
    print(f"{stat.name} 写了 {stat.article_count} 篇文章")

4.3 更新（Update）

# 先查再改，适合需要业务逻辑判断的场景
article = await Article.get(id=1)
article.title = "修改后的标题"
article.is_published = True
await article.save()

# 批量更新（性能更好）
await Article.filter(views__gt=1000).update(is_published=True)

4.4 删除（Delete）

# 单条删除
article = await Article.get(id=1)
await article.delete()

# 批量删除
await Article.filter(is_published=False, created_at__lt="2024-01-01").delete()

5. 集成 FastAPI：5 分钟写完整 REST 接口

将 Pydantic 模型与前面的 CRUD 逻辑结合，一个可立即测试的 API 就诞生了。

from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel, EmailStr
from typing import List
from demo_models import Author, Article, Tag

app = FastAPI(title="Tortoise + FastAPI CRUD 演示")

# ------------------------------ Pydantic 模型 ------------------------------
class AuthorIn(BaseModel):
    name: str
    email: EmailStr

class AuthorOut(BaseModel):
    id: int
    name: str
    email: EmailStr

    class Config:
        from_attributes = True   # 允许直接从 Tortoise 模型转换为 Pydantic

# ------------------------------ 接口实现 ----------------------------------
@app.post("/authors", response_model=AuthorOut, status_code=201)
async def create_author(data: AuthorIn):
    # 尽管模型有 unique 约束，显式检查可以返回更友好的错误信息
    exists = await Author.exists(email=data.email)
    if exists:
        raise HTTPException(status_code=400, detail="该邮箱已被注册")
    return await Author.create(**data.model_dump())

@app.get("/authors", response_model=List[AuthorOut])
async def list_authors(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=50)
):
    return await Author.all().offset(skip).limit(limit)

小提示：以上只是最简示例，实际开发中你可以根据业务需要，添加文章创建、按标签过滤、阅读量更新等更多接口，组合前面学到的 CRUD 方法即可。

6. 总结：速查表与选型建议

6.1 五分钟速查卡

from tortoise import Tortoise

# 初始化（主程序入口）
await Tortoise.init(config=TORTOISE_ORM)
await Tortoise.generate_schemas()          # 首次建表（生产环境用 Aerich）

# 创建
obj = await Model.create(**kwargs)
await Model.bulk_create([Model(...), Model(...)])

# 查询
obj = await Model.get_or_none(id=1)        # 安全版，找不到返回 None
objs = await Model.filter(...)
    .prefetch_related("关联字段")
    .order_by("-时间字段")
    .offset(skip)
    .limit(limit)

# 更新
obj.field = "新值"
await obj.save()
await Model.filter(...).update(field="新值")

# 删除
await obj.delete()
await Model.filter(...).delete()

6.2 什么时候选择 Tortoise-ORM？

✅ 强烈推荐

  • 你喜欢 django ORM 的书写方式，不愿学习新的 ORM
  • 项目为中小型系统、快速原型或 MVP
  • 技术栈基于 FastAPI / Starlette 等异步框架

❌ 谨慎选择

  • 大型企业级系统，需要极度精细的 SQL 控制
  • 已有大量基于 SQLAlchemy 的历史代码
  • 数据库迁移逻辑异常复杂，且需要高度自定义

🔗 扩展资源