qidao123.com技术社区-IT企服评测·应用市场

标题: 异步编程与Tortoise-ORM框架 [打印本页]

作者: 北冰洋以北 时间: 2025-4-19 00:54
标题: 异步编程与Tortoise-ORM框架
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路由中自动管理事件循环
避免在同步代码中直接调用协程

欢迎光临 qidao123.com技术社区-IT企服评测·应用市场 (https://dis.qidao123.com/)