FastAPI + Tortoise ORM 异步数据库实践

引言

去年双十一前，我接手了一个用 FastAPI 写的订单查询服务。压测时 QPS 刚到 800 就上不去了，CPU 却只跑了 30%。打开火焰图一看，asyncio 事件循环里躺着大量 psycopg2 的同步阻塞调用——await db.execute(...) 表面上是异步的，底下却在用线程池硬扛，线程池一满，整个事件循环就开始排队。这就是典型的“伪异步”：你以为自己写的是 async，实际上只是把阻塞操作从主线程挪到了 thread pool。

这个问题的根子不在 FastAPI，而在 ORM 选型。FastAPI 本身是纯异步的 ASGI 框架，但如果你配的是 SQLAlchemy 1.4 的同步 Session，或者 Django ORM，那异步链路就在数据库这一层断掉了。Tortoise ORM 正是为填补这块空缺而生的：它是原生 async 的 ORM，API 设计上高度致敬 Django ORM，让熟悉 Django 的工程师几乎零成本迁移。

这篇文章不讲“怎么装包、怎么定义模型”这种入门内容。我们要往下钻：Tortoise 的连接池是怎么在事件循环里工作的？await Model.get() 背后发生了什么？它和 SQLAlchemy 2.0 async、asyncpg 裸用相比，各自的边界在哪？以及最关键的——生产环境里踩过的那些坑。

---

核心概念：从“餐厅后厨”理解异步 ORM

先把 asyncio 想象成一家只有一个传菜员的餐厅。

• 同步阻塞模型：传菜员把菜端到后厨窗口，然后站在那儿等厨师做完。等的时候他什么也干不了。这就是 psycopg2 的行为。

• 线程池伪异步：老板看传菜员老等着，就雇了 20 个传菜员（线程池）。每个传菜员照样站着等，但至少能同时等 20 单。问题是：传菜员是有成本的（8MB 栈 + 上下文切换），而且 20 个传菜员全占满的时候，第 21 单照样卡住。这就是 SQLAlchemy 1.4 同步 Session 在 async 环境下的真相。

• 真异步：传菜员把订单贴到窗口就转身走人，厨师做完按铃，传菜员再回来取。一个传菜员能同时处理几百单。这就是 Tortoise + asyncpg 的模式——数据库驱动本身是非阻塞的，等待期间事件循环可以调度其他协程。

所以判断一个 ORM 是不是“真异步”，关键看底层的数据库驱动是不是非阻塞的。Tortoise 在 PostgreSQL 下用 asyncpg，MySQL 下用 aiomysql，SQLite 下用 aiosqlite——全部是原生异步驱动，没有线程池兜底。

Tortoise 的另一个设计取舍是放弃 Data Mapper、拥抱 Active Record。SQLAlchemy 是 Data Mapper：模型是纯 Python 对象，和 Session 解耦，Session 负责跟踪状态、生成 SQL。Tortoise 是 Active Record：模型类自带 save()、delete()、filter()，像 Django 一样。这个取舍在异步场景下其实更自然——因为异步里你本来就该“用完即走”，不需要长生命周期的 Session 去维护身份映射（identity map）。这一点后面会细说。

---

源码/原理深度分析

1. 连接池：ConnectionHandler 如何管理事件循环中的连接

Tortoise 的连接管理核心在 tortoise/connection.py 的 ConnectionHandler 类。简化后的关键结构如下：

class ConnectionHandler:
    def __init__(self, connections: dict[str, dict]):
        self._connections: dict[str, ConnectionWrapper] = {}
        self._db_config = connections

    async def _get_connection(self, connection_name: str):
        # 每个连接名对应一个 ConnectionWrapper，
        # 内部持有一个由驱动提供的 pool
        wrapper = self._connections.get(connection_name)
        if wrapper is None:
            # 懒初始化：第一次访问时才建池
            wrapper = ConnectionWrapper(...)
            await wrapper.init(self._db_config[connection_name])
            self._connections[connection_name] = wrapper
        return wrapper

这里有个容易被忽略的细节：连接池是绑定到事件循环的。asyncpg 的 Pool 对象内部持有 asyncio.Queue 和 Loop 引用，如果跨事件循环使用（比如在 uvicorn --workers 4 的多进程里，或者测试中反复 asyncio.run()），就会报 attached to a different loop。我在一个项目里就因为 pytest 每个测试用例都 asyncio.run() 而踩过这个坑。

看 ConnectionWrapper 的 get_connection 实现（tortoise/backends/base/client.py 附近）：

async def get_connection(self):
    if self._pool is None:
        async with self._lock:
            if self._pool is None:
                self._pool = await self.create_pool()
    return await self._pool.acquire()  # 从池里借一个

注意这里用了双检锁（double-checked locking）来保证池只建一次。self._lock 是 asyncio.Lock，在单事件循环里是安全的。这也是为什么 Tortoise 官方明确说不要在多个事件循环之间共享 Tortoise 实例——锁和池都是 loop-bound 的。

2. await Model.get() 的完整生命周期

假设我们执行：

user = await User.get(id=1)

这行代码背后发生的事，可以拆成下面这个流程图：

graph TD A["await User.get(id=1)"] --> B["QuerySet.filter(id=1).first()"] B --> C["构建 Query 对象
解析字段、生成 SQL 与参数"] C --> D["executor.execute_select_query()"] D --> E["ConnectionHandler.get_connection()"] E --> F{"连接池有空闲连接?"} F -->|是| G["acquire() 借出连接"] F -->|否| H["await 等待池中归还
事件循环挂起当前协程"] H --> G G --> I["asyncpg 执行 SQL
期间协程让出 CPU"] I --> J["结果集映射为 model 实例
_init_from_db()"] J --> K["finally: release() 归还连接"] K --> L["返回 User 实例"]

关键点在第 I 步：asyncpg 把 SQL 发给 PostgreSQL 后，当前协程在 await 处挂起，事件循环立刻去跑别的请求。等 PostgreSQL 返回数据，IO 就绪，事件循环再唤醒这个协程继续走 J 和 K。整个过程没有线程参与，也就没有线程池耗尽的问题。

再往深一层看结果集映射。Tortoise 在 Model._init_from_db（tortoise/models.py）里做字段赋值：

@classmethod
def _init_from_db(cls, **kwargs):
    instance = cls.__new__(cls)
    instance._partial = False
    instance._saved_in_db = True
    for key, value in kwargs.items():
        # 直接塞进 __dict__，不做 setter 校验
        instance.__dict__[key] = value
    return instance

注意它绕过了 __setattr__，直接写 __dict__。因为 ORM 的字段描述符（fields.py 里的 Field.__set__）会做类型转换和脏检查，而从数据库读出来的数据已经是可信的，没必要再校验一遍。这是个性能优化，但也意味着——如果你在模型上定义了自定义 __setattr__ 做业务逻辑，从 DB 加载时不会触发。这个坑我见过有人用来做“字段变更审计”，结果读数据时审计日志一条没有。

3. 事务：in_transaction() 与连接的绑定关系

异步事务最容易出错的地方是连接泄漏。Tortoise 的 in_transaction() 是个异步上下文管理器：

async with in_transaction() as conn:
    await User.create(name="a", using_db=conn)
    await Order.create(user_id=..., using_db=conn)

源码层面（tortoise/transactions.py），in_transaction 会从当前连接的池里 acquire 一个连接，BEGIN，然后把事务上下文存进一个 ContextVar：

class _TransactionContext:
    __slots__ = ("connection", "token", "lock")
    ...

async def in_transaction(connection_name=None):
    ctx = _get_context(connection_name)
    await ctx.lock.acquire()
    try:
        connection = await ctx.connection_handler.get_connection()
        await connection.start()
        token = _current_transaction.set(connection)  # ContextVar
        yield connection
    finally:
        await connection.rollback()
        _current_transaction.reset(token)
        ctx.lock.release()

用 ContextVar 而不是 threading.local，是因为 asyncio 的任务切换不换线程。但这也带来一个陷阱：如果你在事务里 await asyncio.create_task(...) 开了子任务，子任务默认继承父任务的 ContextVar 值（Python 3.7+ 的行为），于是子任务里的 ORM 操作会“以为”自己也在同一个事务里，实际上用的是同一个连接——并发执行时会互相踩。正确的做法是子任务里显式传 using_db=None，或者干脆别在事务里开并发子任务。

---

实战代码

下面三个示例都是可运行的完整脚本，聚焦不同层面的实战问题。

示例一：完整 CRUD + 事务 + 连接池配置

# main.py
# 运行：uvicorn main:app --reload
# 依赖：pip install fastapi uvicorn tortoise-orm asyncpg

from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from tortoise import Tortoise, fields, run_async
from tortoise.models import Model
from tortoise.transactions import in_transaction


class User(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=64, index=True)
    balance = fields.DecimalField(max_digits=12, decimal_places=2, default=0)
    created_at = fields.DatetimeField(auto_now_add=True)

    class Meta:
        table = "users"


class Order(Model):
    id = fields.IntField(pk=True)
    user = fields.ForeignKeyField("models.User", related_name="orders")
    amount = fields.DecimalField(max_digits=12, decimal_places=2)
    status = fields.CharEnumField(
        enum_type=type("OrderStatus", (), {"PAID": "paid", "REFUNDED": "refunded"}),
        default="paid",
    )

    class Meta:
        table = "orders"


@asynccontextmanager
async def lifespan(app: FastAPI):
    # 关键：连接池参数直接决定高并发下的表现
    await Tortoise.init(
        db_url="postgres://postgres:postgres@localhost:5432/demo",
        modules={"models": ["main"]},
        # 下面这些参数会透传给 asyncpg 的 create_pool
        connections={
            "default": {
                "engine": "tortoise.backends.asyncpg",
                "credentials": {
                    "host": "localhost",
                    "port": 5432,
                    "user": "postgres",
                    "password": "postgres",
                    "database": "demo",
                    "min_size": 5,      # 池最小连接数
                    "max_size": 20,     # 池上限，别超过 PG 的 max_connections 除以 worker 数
                    "max_queries": 50000,  # 单连接执行多少次查询后回收，防内存泄漏
                    "max_inactive_connection_lifetime": 300,
                },
            }
        },
    )
    await Tortoise.generate_schemas()
    yield
    await Tortoise.close_connections()  # 必须显式关闭，否则测试会挂


app = FastAPI(lifespan=lifespan)


class TransferReq(BaseModel):
    from_user: int
    to_user: int
    amount: float


@app.post("/transfer")
async def transfer(req: TransferReq):
    """转账：演示事务 + 行锁 + 正确回滚"""
    async with in_transaction() as conn:
        # 用 using_db 把两个操作绑到同一连接，
        # 否则 ORM 会从池里再借一个连接，事务就分裂了
        sender = await User.select_for_update().using_db(conn).get_or_none(id=req.from_user)
        if sender is None:
            raise HTTPException(404, "sender not found")
        if sender.balance < req.amount:
            # 抛异常会自动触发 rollback
            raise HTTPException(400, "insufficient balance")

        await User.filter(id=req.from_user).using_db(conn).update(
            balance=sender.balance - req.amount
        )
        await User.filter(id=req.to_user).using_db(conn).update(
            balance=fields.F("balance") + req.amount  # F 表达式，避免读改写竞态
        )
    return {"ok": True}


@app.get("/users/{uid}")
async def get_user(uid: int):
    # prefetch_related 一次查询带出订单，避免 N+1
    user = await User.filter(id=uid).prefetch_related("orders").first()
    if not user:
        raise HTTPException(404)
    return {
        "id": user.id,
        "name": user.name,
        "orders": [{"id": o.id, "amount": str(o.amount)} for o in user.orders],
    }

这段代码里有三个细节值得注意：select_for_update() 要配合 using_db(conn) 才在事务内生效；fields.F() 用于原子更新避免竞态；prefetch_related 是 Tortoise 里避免 N+1 的标准手段。

示例二：批量操作与 bulk_create 的正确姿势

# bulk.py
import asyncio
from tortoise import Tortoise, run_async
from tortoise.expressions import Q
from main import User, Order


async def seed():
    await Tortoise.init(
        db_url="postgres://postgres:postgres@localhost:5432/demo",
        modules={"models": ["main"]},
    )
    await Tortoise.generate_schemas()

    # 反例：循环 create，每次一个 RTT，1 万条要几十秒
    # for i in range(10000):
    #     await User.create(name=f"u{i}")

    # 正例：bulk_create 生成单条 INSERT ... VALUES (...),(...)，一次 RTT
    users = [User(name=f"u{i}") for i in range(10000)]
    await User.bulk_create(users, batch_size=1000)
    # batch_size 控制单条 SQL 的 VALUES 数量，
    # 太大可能撞 PostgreSQL 的 65535 参数上限（每行 N 个字段 × 行数）

    # 批量更新：Tortoise 没有原生 bulk_update，
    # 用 CASE WHEN 或分批 filter().update() 更实际
    await User.filter(name__startswith="u1").update(balance=100)

    # 复杂条件查询：Q 对象组合
    rows = await User.filter(
        Q(name__startswith="u1") & Q(balance__gte=0)
    ).order_by("-created_at").limit(10)
    print(f"matched {len(rows)}")

    await Tortoise.close_connections()


if __name__ == "__main__":
    run_async(seed())

bulk_create 的 batch_size 参数是个双刃剑：调大能减少 RTT，但 PostgreSQL 单条 SQL 最多 65535 个绑定参数，假设 User 有 5 个字段，那 batch_size 超过约 13000 就会报错。默认值 None 表示一次性全塞进去，生产环境务必显式设置。

示例三：并发查询与事件循环陷阱

# concurrency.py
import asyncio
from tortoise import Tortoise
from main import User


async def bad_pattern():
    """反例：在事务里开并发子任务，连接互相踩踏"""
    # 伪代码示意，实际会报 "another operation is in progress"
    # async with in_transaction() as conn:
    #     await asyncio.gather(
    #         User.create(name="a", using_db=conn),
    #         User.create(name="b", using_db=conn),
    #     )
    pass


async def good_pattern():
    """正例：并发查询不共享连接，各自从池里借"""
    await Tortoise.init(
        db_url="postgres://postgres:postgres@localhost:5432/demo",
        modules={"models": ["main"]},
    )
    # 每个查询独立 borrow 连接，事件循环并行调度
    results = await asyncio.gather(
        User.filter(name__startswith="u1").count(),
        User.filter(name__startswith="u2").count(),
        User.filter(balance__gte=50).count(),
    )
    print("counts:", results)

    # 大数据量分页：别用 offset 深翻页，用 keyset pagination
    last_id = 0
    while True:
        page = await User.filter(id__gt=last_id).order_by("id").limit(100)
        if not page:
            break
        for u in page:
            # 处理...
            pass
        last_id = page[-1].id
    await Tortoise.close_connections()


async def main():
    await good_pattern()
    # 注意：不要在一个进程里反复 Tortoise.init / close 后又 init，
    # 池是 loop-bound 的，跨 loop 复用会炸


if __name__ == "__main__":
    asyncio.run(main())

重点在 good_pattern：并发查询必须各自借连接，不能共享。而 bad_pattern 里在事务中 gather 多个写操作，asyncpg 的连接是单工通信，同一时刻只能跑一条语句，会直接抛 another operation is in progress。这是异步 ORM 最经典的坑之一。

---

方案对比：Tortoise vs SQLAlchemy 2.0 async vs asyncpg 裸用

维度	Tortoise ORM	SQLAlchemy 2.0 async	asyncpg 裸用
异步原生性	原生 async 驱动	Session 基于 AsyncSession，底层 asyncpg/aiomysql	原生 async
API 风格	Active Record（类 Django）	Data Mapper（Session 跟踪）	纯 SQL
学习曲线	Django 用户几乎零成本	心智模型重，Session 生命周期管理复杂	最低，但 SQL 全手写
关系加载	prefetch_related 显式	selectinload / joinedload 显式	手写 JOIN
迁移工具	Aerich（功能较基础）	Alembic（成熟）	无
事务隔离	ContextVar 绑定连接	AsyncSession 绑定	手动管理
生态成熟度	中等	极高	高（驱动层）
适用场景	中小型服务、快速迭代、Django 迁移	大型复杂领域模型、多数据库	极致性能、复杂 SQL、只读查询

怎么选？我的经验是分三档：

• 业务 CRUD 为主、团队有 Django 背景：Tortoise 最省心。它的 API 让你几乎不用思考“连接从哪来”，异步链路干净。

• 领域模型复杂、需要多数据源、要成熟迁移：SQLAlchemy 2.0 async。它的 AsyncSession 虽然重，但 selectinload 控制加载策略的能力远超 Tortoise，大型项目后期会庆幸当初选对了。

• 报表、批量导入、复杂聚合：直接 asyncpg。ORM 生成的 SQL 在这种场景下往往是负担，手写 SQL + conn.fetch() 反而更清晰、更快。

还有一个常被忽略的对比维度：Tortoise 的迁移工具 Aerich 相比 Alembic 是明显短板。它不支持自动检测所有 schema 变更（比如索引增删、约束修改），复杂迁移经常要手写 SQL。如果你的项目 schema 会频繁演进，这点要提前评估。

---

最佳实践与避坑指南

1. 连接池参数必须和 PG 的 max_connections 联动算

这是最容易出事的地方。假设 uvicorn --workers 4，每个 worker 一个进程一个事件循环一个池，池 max_size=20，那峰值就是 4×20=80 个连接。PostgreSQL 默认 max_connections=100，再算上运维连接，很快就 too many clients。经验公式：

池 max_size ≤ (PG max_connections - 预留 10) / worker 数

宁可把 max_size 调小（比如 10），靠事件循环的并发能力扛，也别把连接数堆爆。

2. using_db(conn) 一个都不能漏

事务内所有 ORM 操作都必须显式传 using_db=conn。漏一个，那个操作就会从池里另借一个连接，跑在事务之外，回滚时它不跟着回滚——数据就脏了。这是我在 code review 里抓得最多的 bug。

3. 别在事务里 asyncio.gather

前面说过，asyncpg 连接单工，事务内并发写会直接崩。要并发，就在事务外并发，或者拆成多个独立事务。

4. 测试里的 asyncio.run() 是连接池杀手

pytest 每个用例 asyncio.run() 会建新 loop，而 Tortoise 的池绑在旧 loop 上。解决方案是全局一个 loop（pytest-asyncio 的 event_loop fixture），或者在 fixture 里每个用例 Tortoise.init + close_connections。我推荐后者，隔离干净。

5. N+1 查询一定要用 prefetch_related

Tortoise 不会自动预加载。user.orders 第一次访问会触发一次查询，循环里就是 N+1。养成习惯：列表接口返回关联数据时，先 filter(...).prefetch_related("orders")。

6. 深分页用 keyset，别用 offset

offset 100000 limit 20 会让 PG 扫描并丢弃 10 万行。用 filter(id__gt=last_id).order_by("id").limit(20)，走索引，O(1) 定位。

7. Decimal 字段别用 float 中转

balance 是 DecimalField，从 DB 读出来是 Decimal。如果经过 Pydantic 时声明成 float，精度就丢了。序列化时统一 str(amount)，或者 Pydantic 里用 condecimal。

---

总结

回到开头那个 QPS 卡在 800 的服务。换成 Tortoise + asyncpg 后，同样的硬件 QPS 上到 4000+，CPU 也跑满了——因为事件循环终于不用等线程池了。这个提升的本质，不是 ORM 换了个名字，而是数据库 IO 从“线程阻塞”变成了“协程挂起”，一个事件循环能同时挂起成百上千个等待中的查询。

几个要点回顾：

• 异步 ORM 的“异步”取决于底层驱动，Tortoise 用 asyncpg/aiomysql/aiosqlite，是原生异步，没有线程池兜底。

• 连接池是 loop-bound 的，跨事件循环复用会炸，测试和部署都要注意。

• 事务靠 ContextVar 绑定连接，事务内所有操作必须 using_db(conn)，且不能并发共享连接。

• 选型看场景：Tortoise 胜在 Django 式 API 和干净异步链路，SQLAlchemy 2.0 async 胜在复杂模型和成熟迁移，asyncpg 裸用胜在极致控制。

延伸思考：异步 ORM 的下一个战场其实是连接池的可观测性。Tortoise 目前对池的监控指标（等待队列长度、借出/归还速率、慢查询）支持很弱，生产环境经常要靠 pg_stat_activity 反推。如果你的服务规模到了一定量级，考虑在 ConnectionHandler 外面包一层埋点，或者干脆上 PgBouncer 做连接复用——把池的管理从应用层下沉到中间件层，这往往是异步 ORM 高并发场景下的最终形态。