Tortoise-ORM 快速入门：更符合 Python 直觉的异步数据库操作

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

1. 为什么选它？别纠结 SQLAlchemy 了

如果说 SQLAlchemy 是数据库界的「瑞士军刀」，Tortoise-ORM 就是给 Python 开发者定制的「智能折叠剪」——小巧、直观、上手即剪，剪中小型项目/快速原型特别顺。

1.1 一分钟对比 SQLAlchemy 2.0

核心痛点Tortoise-ORM 解法SQLAlchemy 2.0 解法
学习成本完全照搬 Django ORM 的 API 习惯，零额外心智负担2.0 引入 Core + ORM 双轨制，底层概念多，入门得啃1-2周文档
异步体验从出生就是 async/await 原生，没有同步→异步的割裂感2.0 才原生支持异步，但底层还是同步架构优化，需严格遵循异步会话生命周期
配置复杂度一个 Python 字典搞定连接、模型、时区需手动构建异步引擎、异步会话工厂、处理上下文

1.2 一键装完开发环境

选你常用的数据库依赖就行，别全装！

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

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

2. 初始化：写两个文件就能跑

2.1 先搞配置文件 tortoise_config.py

把连接参数、模型模块放在这里，和代码逻辑分开：

from tortoise import Tortoise

TORTOISE_ORM = {
    "connections": {
        "default": {
            "engine": "tortoise.backends.sqlite",
            # "engine": "tortoise.backends.asyncpg",  # 生产用
            "credentials": {
                "file_path": "./dev.db",  # SQLite 本地文件
                # "host": "localhost",
                # "port": 5432,
                # "user": "dev_user",
                # "password": "dev_pwd",
                # "database": "tortoise_demo",
            }
        }
    },
    "apps": {
        # 自定义 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,  # 自动帮你处理不存在、验证失败等异常
)

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="唯一邮箱")
    # 反向关联提示（可选但推荐，方便代码补全）
    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="更新时间")

    # 👇 核心关联关系
    # 外键：正向查作者用 article.author，反向查文章用 author.articles
    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"]  # 新文章在前

4. CRUD 实战：异步操作就是加个 await

所有数据库操作必须加 await，放在异步函数里执行。我们先写几个「纯 Tortoise」的操作示例，再套 FastAPI 路由。

4.1 创建（Create）

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

# 批量创建（性能比循环 create 快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_articles = await Article.all().offset(0).limit(10).order_by("-views")

# 只查指定字段（减少数据传输）
title_list = 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()

# 批量更新：直接用 filter + update，效率高
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分钟写完完整接口

把刚才的操作和 Pydantic 模型绑定，一个可测试的 REST API 就出来了：

from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel, EmailStr
from typing import Optional, 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)

6. 总结：速查表+适用场景

6.1 5分钟速查

# ✅ 核心操作，记住这几个就行
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)  # 安全版
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，不想学新东西
  • 中小型项目、快速原型、MVP
  • 搭配 FastAPI/Starlette 做异步服务

❌ 不推荐场景：

  • 大型企业级系统、需要超精细 SQL 控制
  • 已有复杂的 SQLAlchemy 代码库
  • 数据库需要复杂的自定义迁移

🔗 扩展资源