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

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

1. 为什么选择 Tortoise-ORM？

1.1 对比 SQLAlchemy

特性Tortoise-ORMSQLAlchemy
API 设计更 Pythonic，类似 Django ORM更底层，灵活但学习曲线陡
异步原生异步，开箱即用2.0 才原生支持异步
配置字典式配置，简单直观需要构建引擎和会话
迁移需要配合别的工具Alembic 官方支持
适用中小型项目、快速原型大型企业级项目

1.2 项目依赖

pip install tortoise-orm[asyncpg] asyncpg
# asyncpg → PostgreSQL（生产）
# aiosqlite → SQLite（开发）：pip install tortoise-orm[sqlite]

2. 配置与初始化

2.1 基础配置

# tortoise_config.py
from tortoise import Tortoise, fields
from tortoise.fields import ForeignKeyField, ManyToManyField

TORTOISE_ORM = {
    "connections": {
        "default": {
            "engine": "tortoise.backends.sqlite",      # 开发
            # "engine": "tortoise.backends.asyncpg",   # 生产用 PostgreSQL
            "credentials": {
                "file_path": "./data.db",              # SQLite 文件路径
                # "host": "localhost",
                # "port": 5432,
                # "user": "admin",
                # "password": "secret",
                # "database": "mydb",
            }
        }
    },
    "apps": {
        "models": {
            "models": ["__main__", "models.article"],  # 模型所在模块
            "default_connection": "default",
        }
    },
    "use_tz": False,
    "timezone": "Asia/Shanghai",
}

2.2 在 FastAPI 中初始化

# main.py
from fastapi import FastAPI
from tortoise.contrib.fastapi import register_tortoise

app = FastAPI()

# 方式一：register_tortoise 自动注册
register_tortoise(
    app,
    db_url="sqlite://./data.db",
    modules={"models": ["__main__", "models.article"]},
    generate_schemas=True,          # 首次运行自动建表
    add_exception_handlers=True,     # 自动添加 CRUD 异常处理
)

# 方式二：手动初始化（更灵活）
@app.on_event("startup")
async def init():
    await Tortoise.init(config=TORTOISE_ORM)
    await Tortoise.generate_schemas()

@app.on_event("shutdown")
async def close():
    await Tortoise.close_connections()

3. 模型定义

3.1 模型基础

from tortoise import fields
from tortoise.models import Model

class Category(Model):
    id = fields.IntField(pk=True, autoincrement=True)
    name = fields.CharField(max_length=50, unique=True)
    description = fields.TextField(null=True)
    created_at = fields.DatetimeField(auto_now_add=True)

    class Meta:
        table = "categories"

    def __str__(self):
        return self.name


class Article(Model):
    id = fields.IntField(pk=True)
    title = fields.CharField(max_length=200)
    content = fields.TextField()
    views = fields.IntField(default=0)
    published = fields.BooleanField(default=False)
    created_at = fields.DatetimeField(auto_now_add=True)
    updated_at = fields.DatetimeField(auto_now=True)

    # 关联关系
    author: fields.ForeignKeyRelation["Author"] = fields.ForeignKeyField(
        "models.author", related_name="articles"
    )
    tags: fields.ManyToManyRelation["Tag"] = fields.ManyToManyField(
        "models.tag", related_name="articles", through="article_tag"
    )

    class Meta:
        table = "articles"
        ordering = ["-created_at"]

    def __str__(self):
        return self.title


class Tag(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=30, unique=True)
    articles: fields.ManyToManyRelation[Article]

    def __str__(self):
        return self.name

3.2 支持的字段类型

字段说明示例
IntField整数id = fields.IntField(pk=True)
CharField字符串name = fields.CharField(max_length=100)
TextField长文本content = fields.TextField()
BoolField布尔active = fields.BooleanField(default=True)
DatetimeField日期时间created = fields.DatetimeField(auto_now_add=True)
DateField日期birthday = fields.DateField()
FloatField浮点数price = fields.FloatField()
DecimalField精确小数price = fields.DecimalField(max_digits=10, decimal_places=2)
JSONFieldJSON 数据config = fields.JSONField()
ForeignKeyField外键author = fields.ForeignKeyField("models.Author")
ManyToManyField多对多tags = fields.ManyToManyField("models.Tag")

4. CRUD 操作

4.1 创建（Create）

# 创建单条
article = await Article.create(
    title="FastAPI 入门",
    content="这是一篇关于 FastAPI 的教程...",
    author_id=1,
)
article.published = True
await article.save()

# 批量创建
await Article.bulk_create([
    Article(title="文章1", content="内容1", author_id=1),
    Article(title="文章2", content="内容2", author_id=1),
])

# 或使用字典创建
await Article.create(title="文章3", content="内容3", author_id=1)

4.2 读取（Read）

# 查询单条
article = await Article.get(id=1)
article = await Article.get_or_none(id=999)  # 不存在返回 None，不会抛异常

# 查询多条
articles = await Article.filter(published=True)
articles = await Article.filter(author_id=1, published=True)
articles = await Article.filter(tags__name="Python")  # 跨关联查询

# 分页
articles = await Article.all().offset(0).limit(10)

# 排序
articles = await Article.all().order_by("-created_at")  # 降序

# 预加载关联
articles = await Article.all().prefetch_related("author", "tags")

# 使用 .only() 指定字段
titles = await Article.all().only("id", "title")

# 统计
count = await Article.filter(published=True).count()

# 是否存在
exists = await Article.exists(id=1)

# 聚合
from tortoise.functions import Count, Max, Avg
from tortoise.aggregations import Sum

# 按作者统计文章数
from tortoise.models import Q
stats = await Article.all().annotate(
    article_count=Count("id"),
).group_by("author_id")

4.3 更新（Update）

# 单条更新
article = await Article.get(id=1)
article.title = "新标题"
await article.save()

# 批量更新
await Article.filter(published=False).update(published=True)

# 使用 Q 对象组合条件
from tortoise.models import Q
await Article.filter(Q(views__gt=100) | Q(published=True)).update(hot=True)

4.4 删除（Delete）

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

# 批量删除
await Article.filter(published=False).delete()

5. 关联查询

5.1 外键查询

# 正向查询（从文章查作者）
article = await Article.get(id=1).prefetch_related("author")
print(article.author.name)

# 反向查询（从作者查所有文章）
author = await Author.get(id=1).prefetch_related("articles")
for article in author.articles:
    print(article.title)

5.2 多对多查询

# 给文章添加标签
article = await Article.get(id=1)
tag = await Tag.get(name="Python")
await article.tags.add(tag)

# 获取文章的所有标签
article = await Article.get(id=1).prefetch_related("tags")
for tag in article.tags:
    print(tag.name)

# 获取某标签下的所有文章
tag = await Tag.get(name="Python").prefetch_related("articles")
for article in tag.articles:
    print(article.title)

# 移除关联
await article.tags.remove(tag)

6. Tortoise-ORM 专用路由依赖

6.1 注入当前用户模型

from tortoise.contrib.fastapi import tortoise_app_context

# Tortoise ORM 的中间件，为每个请求创建数据库连接
@app.middleware("http")
async def db_context_middleware(request: Request, call_next):
    async with tortoise_app_context():
        response = await call_next(request)
    return response

6.2 完整 CRUD 路由示例

from fastapi import FastAPI, HTTPException, Query
from tortoise.contrib.fastapi import HTTPNotFoundError
from pydantic import BaseModel
from typing import Optional, List

app = FastAPI()

# Pydantic Schema
class ArticleIn(BaseModel):
    title: str
    content: str
    author_id: int

class ArticleOut(BaseModel):
    id: int
    title: str
    content: str
    author_id: int

    class Config:
        from_attributes = True

# ── CRUD 路由 ────────────────────────────────────
@app.post("/articles", response_model=ArticleOut, status_code=201)
async def create_article(data: ArticleIn):
    article = await Article.create(**data.model_dump())
    return article

@app.get("/articles", response_model=List[ArticleOut])
async def list_articles(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100),
    published: Optional[bool] = None,
):
    query = Article.all()
    if published is not None:
        query = query.filter(published=published)
    articles = await query.offset(skip).limit(limit).order_by("-created_at")
    return articles

@app.get("/articles/{article_id}", response_model=ArticleOut)
async def get_article(article_id: int):
    article = await Article.get_or_none(id=article_id)
    if not article:
        raise HTTPException(status_code=404, detail="文章不存在")
    return article

@app.put("/articles/{article_id}", response_model=ArticleOut)
async def update_article(article_id: int, data: ArticleIn):
    article = await Article.get_or_none(id=article_id)
    if not article:
        raise HTTPException(status_code=404, detail="文章不存在")
    await article.update_from_dict(data.model_dump())
    await article.save()
    return article

@app.delete("/articles/{article_id}", status_code=204)
async def delete_article(article_id: int):
    article = await Article.get_or_none(id=article_id)
    if not article:
        raise HTTPException(status_code=404, detail="文章不存在")
    await article.delete()

7. 小结

# Tortoise-ORM 五分钟速查

# 初始化
await Tortoise.init(config=TORTOISE_ORM)
await Tortoise.generate_schemas()

# 创建
obj = await Model.create(field=value)

# 查询
obj = await Model.get(id=1)
objs = await Model.filter(field=value).prefetch_related("relation")

# 更新
obj.field = new_value
await obj.save()
await Model.filter(id=1).update(field=new_value)

# 删除
await obj.delete()
await Model.filter(id=1).delete()

💡 选择建议：如果你喜欢 Django ORM 的风格、需要快速开发中小型项目，Tortoise-ORM 是绝佳选择。如果你在做大型企业级系统、需要精细的 SQL 控制，SQLAlchemy 2.0 更合适。

🔗 扩展阅读