title: 异步IO与Tortoise-ORM的数据库
date: 2025/04/29 13:21:47
updated: 2025/04/29 13:21:47
author: cmdragon
excerpt:
异步IO与同步IO的核心区别在于阻塞与非阻塞模式。Tortoise-ORM通过协议层、连接池层和ORM层实现异步数据库操纵，支持高效的并发处理。用户管理系统搭建中，Tortoise-ORM与FastAPI结合，实现了用户创建和查询功能，并通过Pydantic进行数据校验。异步ORM适用于高并发场景，参数化查询可防止SQL注入。最佳实践包括连接池设置、查询优化和事件管理，确保系统性能和数据一致性。
categories:
tags:
- 异步IO
- Tortoise-ORM
- 数据库操纵
- FastAPI
- 异步编程
- 连接池
- 事件管理
扫描二维码
关注或者微信搜一搜：编程智域 前端至全栈交换与发展
探索数千个预构建的 AI 应用，开启你的下一个伟大创意：https://tools.cmdragon.cn/
第一章：异步IO与Tortoise-ORM原理剖析
1.1 同步与异步的本质区别
想象你在快餐店点餐：
- 同步模式：收银员接单后站在炸薯条机前等待，直到薯条炸好才接待下一位顾客
- 异步模式：收银员接单后立即将订单交给后厨，转身接待下一位顾客，后厨预备好餐点会主动通知收银员
盘算机范畴的异步IO正是采用这种"非阻塞"模式：- # 同步操作（线程阻塞）
- def sync_query():
- result = db.execute("SELECT * FROM users") # 线程在此等待
- process(result)
- # 异步操作（事件驱动）
- async def async_query():
- result = await db.execute("SELECT * FROM users") # 释放控制权
- process(result)
Tortoise-ORM通过三层架构实现异步操纵：
层级职责关键技能协议层数据库通讯协议解析asyncpg/aiomysql连接池层管理异步数据库连接asyncio.QueueORM层模型映射与查询构建Python元类编程典型查询流程解析：- async def get_users():
- # 以下三个步骤交替执行，全程无阻塞
- users = await User.filter(age__gt=18) # 1.生成SQL语句
- # 2.从连接池获取连接
- # 3.等待数据库响应
- return users
情况预备
- pip install fastapi uvicorn tortoise-orm aiosqlite pydantic
- project/
- ├── config.py
- ├── models.py
- ├── schemas.py
- └── main.py
- from tortoise.models import Model
- from tortoise import fields
- class User(Model):
- id = fields.IntField(pk=True)
- username = fields.CharField(max_length=50, unique=True)
- hashed_password = fields.CharField(max_length=128)
- email = fields.CharField(max_length=100)
- created_at = fields.DatetimeField(auto_now_add=True)
- class Meta:
- table = "users"
- from pydantic import BaseModel, EmailStr
- class UserCreate(BaseModel):
- username: str
- password: str
- email: EmailStr
- class Config:
- schema_extra = {
- "example": {
- "username": "fastapi_user",
- "password": "strongpassword123",
- "email": "user@example.com"
- }
- }
- from fastapi import FastAPI, Depends, HTTPException
- from tortoise.contrib.fastapi import register_tortoise
- from models import User
- from schemas import UserCreate
- app = FastAPI()
- # 初始化数据库
- register_tortoise(
- app,
- db_url="sqlite://db.sqlite3",
- modules={"models": ["models"]},
- generate_schemas=True,
- add_exception_handlers=True,
- )
- @app.post("/users/", status_code=201)
- async def create_user(user_data: UserCreate):
- # 密码哈希处理（实际项目应使用passlib）
- hashed_password = f"hashed_{user_data.password}"
- try:
- user = await User.create(
- username=user_data.username,
- hashed_password=hashed_password,
- email=user_data.email
- )
- except Exception as e:
- raise HTTPException(
- status_code=400,
- detail="Username already exists"
- )
- return {
- "id": user.id,
- "username": user.username,
- "email": user.email
- }
- @app.get("/users/{user_id}")
- async def get_user(user_id: int):
- user = await User.get_or_none(id=user_id)
- if not user:
- raise HTTPException(status_code=404, detail="User not found")
- return {
- "id": user.id,
- "username": user.username,
- "email": user.email,
- "created_at": user.created_at.isoformat()
- }
题目1：以下哪种场景最适合使用异步ORM？
A) 单用户的桌面应用程序
B) 需要处理数千并发哀求的API服务
C) 执行复杂事件的财务系统
D) 数据堆栈的批量数据处理
答案：B
解析：异步ORM在高并发IO密集型场景下能明显提升吞吐量，而ACD场景更多需要的是事件完整性或盘算能力。
题目2：如何避免在ORM查询时发生SQL注入？
A) 直接拼接字符串
B) 使用ORM的参数化查询
C) 手动过滤特殊字符
D) 限制查询字段长度
答案：B
解析：Tortoise-ORM的查询方法会主动进行参数化处理，有效防止SQL注入，这是最安全的做法。
常见报错解决方案
错误1：422 Validation Error
原因分析：哀求体不符合Pydantic模型要求
解决方法：
- 检查哀求头Content-Type是否为application/json
- 使用Swagger文档测试接口
- 查看返回信息中的错误字段提示
错误2：RuntimeError: Event loop is closed
原因分析：异步代码在错误的位置执行
解决方法：
- 确保所有异步操纵都在async函数内
- 使用asyncio.run()正确启动事件循环
- 检查数据库连接是否正确关闭
错误3：OperationalError: Connection refused
原因分析：数据库连接设置错误
解决方法：
- 检查db_url格式：dialect://user:password@host:port/database
- 确认数据库服务是否正常运行
- 验证网络防火墙设置
最佳实践发起
- 连接池设置：根据数据库最大连接数设置maxsize
- register_tortoise(
- app,
- db_url="postgres://user:pass@localhost:5432/mydb",
- modules={"models": ["models"]},
- generate_schemas=True,
- add_exception_handlers=True,
- connection_params={
- "maxsize": 20 # 控制连接池大小
- }
- )
- 查询优化：使用select_related预加载关联数据
- # 获取用户及其所有文章
- async def get_user_with_posts(user_id: int):
- user = await User.get(id=user_id).prefetch_related('posts')
- return user
- async def transfer_funds(from_id, to_id, amount):
- async with in_transaction() as conn:
- from_user = await User.get(id=from_id).for_update()
- to_user = await User.get(id=to_id).for_update()
- if from_user.balance < amount:
- raise ValueError("Insufficient balance")
- from_user.balance -= amount
- to_user.balance += amount
- await from_user.save(using_db=conn)
- await to_user.save(using_db=conn)
往期文章归档：
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。 |
