title: 异步编程与Tortoise-ORM框架
date: 2025/04/19 00:13:05
updated: 2025/04/19 00:13:05
author: cmdragon

excerpt:
异步编程通过async/await语法实现协程，单线程可处理多个并发请求，适合IO密集型场景。Tortoise-ORM专为异步设计，支持完整ORM功能和多种数据库，与Pydantic深度集成。整合FastAPI时，通过register_tortoise初始化ORM，使用in_transaction管理事务，确保操作原子性。常见问题包括未使用await返回协程对象和事件循环关闭错误，需通过正确的事件循环启动和事务管理解决。

categories:

后端开发
FastAPI

tags:

异步编程
Tortoise-ORM
FastAPI
协程机制
数据库事务
Pydantic集成
异步IO

扫描二维码关注或者微信搜一搜：编程智域前端至全栈交流与成长

探索数千个预构建的 AI 应用，开启你的下一个伟大创意

第一章：异步编程基础与Tortoise-ORM框架定位

1.1 异步IO原理与协程机制

当你在快餐店排队时，同步IO就像站在队列中干等取餐，而异步IO则是先下单后去玩手机，听到叫号再取餐。现代Web应用需要同时服务成千上万个这样的"
顾客"，这正是异步编程的价值所在。

Python通过async/await语法实现协程：

async def fetch_data():
    # 模拟IO操作
    await asyncio.sleep(1)
    return {"data": "result"}


# 事件循环驱动执行
async def main():
    task1 = fetch_data()
    task2 = fetch_data()
    await asyncio.gather(task1, task2)  # 并发执行


asyncio.run(main())

关键点解析：

async def 声明异步函数（协程）
await 将控制权交还事件循环
单个线程可处理多个并发请求

与传统同步模型对比：

指标	同步模式	异步模式
线程使用	1请求1线程	单线程处理多请求
IO等待处理	阻塞	非阻塞
适合场景	CPU密集型	IO密集型

1.2 Tortoise-ORM的异步设计哲学

传统ORM（如Django ORM）在异步环境中会形成性能瓶颈。Tortoise-ORM专为异步而生，其架构设计呈现以下特点：

from tortoise.models import Model
from tortoise import fields


class User(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=50)
    created_at = fields.DatetimeField(auto_now_add=True)

    class Meta:
        table = "auth_user"

框架核心优势：

完整的ORM功能支持（关系、事务、聚合）
原生异步查询接口设计
支持PostgreSQL/MySQL/SQLite
与Pydantic深度集成

1.3 整合FastAPI的完整示例

创建具备完整功能的API端点：

from fastapi import FastAPI, Depends
from tortoise.contrib.fastapi import register_tortoise
from pydantic import BaseModel

app = FastAPI()


# 请求体模型
class UserCreate(BaseModel):
    name: str


# 响应模型
class UserOut(UserCreate):
    id: int
    created_at: datetime


# 数据库配置
DB_CONFIG = {
    "connections": {"default": "sqlite://db.sqlite3"},
    "apps": {
        "models": {
            "models": ["__main__"],  # 自动发现当前模块的模型
            "default_connection": "default",
        }
    },
}

# 注册Tortoise-ORM
register_tortoise(
    app,
    config=DB_CONFIG,
    generate_schemas=True,  # 自动建表
    add_exception_handlers=True,
)


# 依赖注入数据库连接
async def get_db():
    async with in_transaction() as conn:
        yield conn


@app.post("/users", response_model=UserOut)
async def create_user(user: UserCreate, conn=Depends(get_db)):
    """
    创建用户并返回完整数据
    使用事务保证原子性操作
    """
    db_user = await User.create(**user.dict(), using_db=conn)
    return UserOut.from_orm(db_user)

代码要点解析：

register_tortoise 实现ORM初始化
in_transaction 管理事务作用域
using_db 参数确保使用同一连接
from_orm 自动转换模型为Pydantic对象

课后Quiz

Q1：当数据库查询未使用await时会导致什么现象？
A. 立即返回查询结果
B. 抛出RuntimeWarning
C. 返回coroutine对象
D. 程序崩溃

正确答案：C
解析：异步函数必须使用await执行，否则将返回未被执行的协程对象，这是常见的初学者错误。

Q2：如何确保多个更新操作在同一个事务中？
A. 使用@transaction装饰器
B. 手动begin/commit
C. 通过in_transaction上下文管理器
D. 所有操作自动在事务中

正确答案：C
解析：async with in_transaction() as conn会创建事务作用域，所有在该上下文中的操作使用同一个连接。

常见报错解决方案

问题1：422 Unprocessable Entity

{
  "detail": [
    {
      "loc": [
        "body",
        "name"
      ],
      "msg": "field required",
      "type": "value_error.missing"
    }
  ]
}

原因分析：

请求体缺少name字段
字段类型不匹配（如数字传字符串）
Pydantic模型校验失败

解决方案：

检查请求体是否符合API文档
使用Swagger UI进行测试
查看模型字段定义是否包含required=True

问题2：RuntimeError: Event loop is closed
产生场景：

# 错误写法
async def get_data():
    await User.all()


# 同步上下文中直接调用
get_data()

正确处理：

async def main():
    await get_data()


if __name__ == "__main__":
    import asyncio

    asyncio.run(main())

预防建议：

始终通过事件循环启动异步程序
在FastAPI路由中自动管理事件循环
避免在同步代码中直接调用协程

余下文章内容请点击跳转至个人博客页面或者扫码关注或者微信搜一搜：编程智域前端至全栈交流与成长，阅读完整的文章：异步编程与Tortoise-ORM框架 | cmdragon's Blog