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.dbsqlite://: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_nowauto_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 事件中初始化,确保应用关闭时释放连接:始终在 lifespanyield 后调用 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 = "./."

参见

阅读更多

Web 安全基础

1. HTML 转义(服务端渲染必须): 2. CSP(Content Security Policy): 3. HttpOnly Cookie:防止 JS 读取会话 Cookie: 4. 前端框架防护: 攻击者在第三方网站构造一个表单,诱导已登录用户提交,浏览器会自动携带目标站的 Cookie。 触发条件: 1. 用户已登录目标网站(Cookie 有效) 2. 目标 API 仅凭 Cookie 识别用户身份 3. 请求来源未验证 1. CSRF Token(推荐): 2. SameSite Cookie: 3. 验证 Origin/Referer 头:

By yellowdog

HTTP 协议深度指南

HTTP(HyperText Transfer Protocol)是 Web 的基础传输协议,基于 TCP/IP,采用请求/响应模型。 相关文档:Web安全基础(/web-an-quan-ji-chu/) FastAPI完全指南(/fastapi-wan-quan-zhi-nan/) Nginx完全指南(/nginx-wan-quan-zhi-nan/) 幂等性:多次执行相同请求,服务器状态结果相同。PUT /users/1 多次执行结果一致;POST /users 每次创建新资源,非幂等。 浏览器直接从本地缓存读取,不向服务器发送请求。 缓存命中时,状

By yellowdog

系统设计基础

SLA 对照表: 选择建议:无状态服务(Web 层、API 层)优先水平扩展;数据库初期垂直扩展,达到瓶颈后考虑分库分表或读写分离。 缓存穿透(查询不存在的 key,每次都打到 DB): 缓存击穿(热点 key 过期,瞬间大量请求打到 DB): 缓存雪崩(大量 key 同时过期,或缓存服务宕机): 令牌桶 Python 实现: Redis 实现分布式限流(滑动窗口): URL 命名规则: Cursor 分页响应格式: 雪花算法结构(64 bit): 定义:分布式系统不能同时满足以下三个特性: 在分布式环境中 P 是必须保证的,所以实际是 CP vs AP

By yellowdog

算法思路与模板

二分查找要求序列有序,每次将搜索范围缩减一半,时间复杂度 O(log n)。 两个指针从两端向中间收缩,常用于有序数组。 滑动窗口维护一个满足条件的区间 left, right,right 不断向右扩张,条件不满足时收缩 left。 滑动窗口通用框架: 1. 确定"子问题":原问题可以分解为哪些规模更小的同类问题 2. 定义 dpi 或 dpij 的含义,要足够清晰 3. 推导状态转移方程 4. 确定初始状态(边界条件) 5. 确定计算顺序(确保依赖的子问题先计算) 每件物品最多选一次。dpj = 容量为 j 时的最大价值,逆序遍历容量防止重复选取。 每

By yellowdog