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

探索数千个预构建的 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)

1.2 Tortoise-ORM的异步实现

Tortoise-ORM通过三层架构实现异步操作：

层级	职责	关键技术
协议层	数据库通信协议解析	asyncpg/aiomysql
连接池层	管理异步数据库连接	asyncio.Queue
ORM层	模型映射与查询构建	Python元类编程

典型查询流程解析：

async def get_users():
    # 以下三个步骤交替执行，全程无阻塞
    users = await User.filter(age__gt=18)  # 1.生成SQL语句
    # 2.从连接池获取连接
    # 3.等待数据库响应
    return users

1.3 实战：用户管理系统搭建

环境准备

1	pip install fastapi uvicorn tortoise-orm aiosqlite pydantic

项目结构

project/
├── config.py
├── models.py
├── schemas.py
└── main.py

模型定义（models.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"

数据校验（schemas.py）

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"
            }
        }

核心逻辑（main.py）

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()
    }

课后Quiz

问题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)

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

异步IO与Tortoise-ORM的数据库