Tortoise ORM 初始化与配置
pyproject.toml 配置: pydantic_model_creator 参数: .env 文件: 将数据库配置单独放在 config.py 或环境变量中,不要硬编码在 main.py:URL 中含有密码,应通过 os.environ 或 .env 文件注入,避免泄露到版本控制。 使用 aerich 管理迁移,不要在生产环境调用 generate_schemas():generate_schemas() 只能新建表,无法处理字段变更。生产环境应用 aerich upgrade 执行迁移文件,保证数据不丢失。 lifespan 事件中初始化,确保
官方文档:https://tortoise.github.io/getting_started.html
适用版本:tortoise-orm >= 0.20(2026-05-07 核实)
一、安装
pip install tortoise-orm
# 按数据库选择驱动
pip install tortoise-orm[asyncpg] # PostgreSQL
pip install tortoise-orm[aiomysql] # MySQL
pip install tortoise-orm[aiosqlite] # SQLite(开发用)
# 数据库迁移工具(推荐)
pip install aerich
二、Tortoise.init() 参数详解
from tortoise import Tortoise
await Tortoise.init(
db_url='postgres://user:pass@localhost:5432/dbname',
modules={'models': ['myapp.models']},
)
await Tortoise.generate_schemas() # 生成/同步表结构
完整参数
| 参数 | 类型 | 说明 |
|---|---|---|
config |
dict |
完整配置字典,与 db_url + modules 二选一 |
config_file |
str |
配置文件路径(JSON/YAML),与 config 二选一 |
db_url |
str |
数据库连接 URL(简洁写法) |
modules |
dict |
应用名 → 模型模块列表 |
routers |
list |
自定义数据库路由器 |
use_tz |
bool |
是否使用时区感知 datetime(推荐 True) |
timezone |
str |
时区名称,如 'Asia/Shanghai'(需要 use_tz=True) |
db_url 格式
{engine}://{user}:{password}@{host}:{port}/{dbname}?{params}
| 数据库 | 示例 |
|---|---|
| SQLite | sqlite://./dev.db 或 sqlite://:memory: |
| PostgreSQL | postgres://user:pass@localhost:5432/mydb |
| MySQL | mysql://user:pass@localhost:3306/mydb |
完整 config 写法
TORTOISE_ORM = {
'connections': {
'default': {
'engine': 'tortoise.backends.asyncpg',
'credentials': {
'host': 'localhost',
'port': 5432,
'user': 'myuser',
'password': 'mypass',
'database': 'mydb',
'minsize': 1, # 连接池最小连接数
'maxsize': 10, # 连接池最大连接数
}
}
},
'apps': {
'models': {
'models': ['myapp.models', 'aerich.models'],
'default_connection': 'default',
}
},
'use_tz': True,
'timezone': 'Asia/Shanghai',
}
await Tortoise.init(config=TORTOISE_ORM)
三、generate_schemas() 参数
await Tortoise.generate_schemas(safe=True)
| 参数 | 默认 | 说明 |
|---|---|---|
safe |
False |
True:表已存在时不报错(CREATE TABLE IF NOT EXISTS) |
开发环境:
safe=True方便调试。
生产环境:不要用generate_schemas,使用 Aerich 做版本化迁移。
四、FastAPI 完整集成
# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from tortoise import Tortoise
from tortoise.contrib.fastapi import RegisterTortoise
TORTOISE_CONFIG = {
'connections': {'default': 'postgres://user:pass@localhost/mydb'},
'apps': {
'models': {
'models': ['myapp.models', 'aerich.models'],
'default_connection': 'default',
}
},
'use_tz': True,
'timezone': 'Asia/Shanghai',
}
# 方法一:lifespan
@asynccontextmanager
async def lifespan(app: FastAPI):
await Tortoise.init(config=TORTOISE_CONFIG)
yield
await Tortoise.close_connections()
app = FastAPI(lifespan=lifespan)
# 方法二:RegisterTortoise(tortoise-orm 官方提供)
app = FastAPI()
RegisterTortoise(
app,
config=TORTOISE_CONFIG,
generate_schemas=True, # 仅开发使用
add_exception_handlers=True, # 添加 DoesNotExist 等异常处理
)
五、Aerich 迁移工具
# 初始化
aerich init -t myapp.config.TORTOISE_ORM
# 初始化数据库
aerich init-db
# 创建迁移
aerich migrate --name add_user_table
# 应用迁移
aerich upgrade
# 回滚
aerich downgrade
# 查看历史
aerich history
pyproject.toml 配置:
[tool.aerich]
tortoise_orm = "myapp.config.TORTOISE_ORM"
location = "./migrations"
src_folder = "./."
六、Pydantic 集成(API 序列化)
from tortoise.contrib.pydantic import pydantic_model_creator
from myapp.models import User
# 生成 Pydantic 模型
UserOut = pydantic_model_creator(User, name='UserOut', exclude=('password_hash',))
UserIn = pydantic_model_creator(User, name='UserIn', include=('username', 'email', 'password_hash'))
# 在路由中使用
@app.get('/users/{user_id}', response_model=UserOut)
async def get_user(user_id: int):
user = await User.get(id=user_id)
return await UserOut.from_tortoise_orm(user) # 转换为 Pydantic 模型(自动预取关联)
@app.get('/users', response_model=list[UserOut])
async def list_users():
return await UserOut.from_queryset(User.all())
# from_queryset_single:传入单对象 QuerySet(如 get() 返回值),等同于 from_tortoise_orm 但接受 QuerySetSingle
@app.get('/users/{user_id}/alt', response_model=UserOut)
async def get_user_alt(user_id: int):
return await UserOut.from_queryset_single(User.get(id=user_id))
pydantic_model_creator 参数:
| 参数 | 默认值 | 说明 |
|---|---|---|
cls |
✅ | Tortoise 模型类 |
name |
None |
Pydantic 模型名称;None 则使用模型类名 |
exclude |
None |
排除的字段名元组 |
include |
None |
只包含的字段名元组 |
exclude_readonly |
False |
排除只读字段(auto_now、auto_now_add 等) |
optional |
None |
标记为可选的字段名元组 |
computed |
None |
额外计算属性名元组 |
allow_cycles |
None |
是否允许循环引用 |
sort_alphabetically |
None |
是否按字母顺序排序字段 |
meta_override |
None |
覆盖自动生成的 Pydantic Meta 配置类 |
model_config |
None |
直接传入 Pydantic ConfigDict 覆盖模型配置 |
validators |
None |
额外的 Pydantic 验证器 |
七、环境变量最佳实践
# config.py
import os
from dotenv import load_dotenv
load_dotenv()
DATABASE_URL = os.getenv('DATABASE_URL', 'sqlite://./dev.db')
TORTOISE_ORM = {
'connections': {'default': DATABASE_URL},
'apps': {
'models': {
'models': ['myapp.models', 'aerich.models'],
'default_connection': 'default',
}
},
'use_tz': True,
'timezone': 'Asia/Shanghai',
}
.env 文件:
DATABASE_URL=postgres://user:pass@localhost:5432/mydb
八、测试配置
# conftest.py
import pytest
from tortoise import Tortoise
from tortoise.contrib.test import initializer, finalizer
@pytest.fixture(scope='session')
def anyio_backend():
return 'asyncio'
@pytest.fixture(autouse=True)
async def db():
await Tortoise.init(
db_url='sqlite://:memory:', # 内存数据库,测试结束自动销毁
modules={'models': ['myapp.models']},
)
await Tortoise.generate_schemas()
yield
await Tortoise.close_connections()
# 或用官方 fixture
from tortoise.contrib.pytest import tortoise_orm_config
# pytest.ini
# [pytest]
# asyncio_mode = auto
最佳实践
将数据库配置单独放在 config.py 或环境变量中,不要硬编码在 main.py:URL 中含有密码,应通过 os.environ 或 .env 文件注入,避免泄露到版本控制。
import os
TORTOISE_ORM = {
'connections': {
'default': os.environ['DATABASE_URL'],
},
'apps': {
'models': {'models': ['myapp.models', 'aerich.models'], 'default_connection': 'default'}
},
}
使用 aerich 管理迁移,不要在生产环境调用 generate_schemas():generate_schemas() 只能新建表,无法处理字段变更。生产环境应用 aerich upgrade 执行迁移文件,保证数据不丢失。
lifespan 事件中初始化,确保应用关闭时释放连接:始终在 lifespan 的 yield 后调用 Tortoise.close_connections(),避免应用关闭时出现 "connection already closed" 警告或连接池泄漏。
@asynccontextmanager
async def lifespan(app):
await Tortoise.init(config=TORTOISE_ORM)
yield
await Tortoise.close_connections() # 必须
多应用拆分时,为每个 app 单独声明 models 列表:避免将所有模型堆在同一个 app 中,便于迁移文件隔离和代码复用。
配置 timezone 为 UTC:Tortoise ORM 在存储 DatetimeField 时依赖数据库和应用层的时区一致性。统一使用 UTC 存储,业务层按需转换,避免夏令时和时区切换引起的数据错误。
常见陷阱
陷阱:在 FastAPI 路由处理函数外调用 ORM 操作导致 "No connections"
现象: 在模块顶层(import 阶段)执行 await Model.all() 时,抛出 ConfigurationError: No DB connections。
原因: Tortoise ORM 需要先调用 Tortoise.init() 建立连接,而 init() 是在 lifespan 事件(应用启动时)执行的,此时模块已被导入,但 DB 连接尚未建立。
解决: 所有 ORM 调用必须在路由函数或 lifespan yield 之后执行,不能在模块顶层调用。
陷阱:generate_schemas() 在生产环境导致字段变更丢失
现象: 修改了模型字段(如新增列、修改类型),但数据库中的表结构没有变化;或在某些数据库上 generate_schemas() 报 "Table already exists"。
原因: generate_schemas() 只创建不存在的表,对已存在的表不做任何修改,不能替代数据库迁移。
解决: 使用 aerich 管理迁移,生产环境通过 aerich upgrade 应用变更。
陷阱:多个 apps 共用同一迁移目录导致冲突
现象: aerich init-db 后运行 aerich migrate 报错,或不同 app 的迁移文件互相覆盖。
原因: 每个 aerich app 需要独立的迁移目录,多个 app 共用同一目录时文件命名冲突。
解决: 在 pyproject.toml 中为每个 app 配置独立的 location。
[tool.aerich]
tortoise_orm = "myapp.config.TORTOISE_ORM"
location = "./migrations/main"
src_folder = "./."