13.4 异步数据库操作最佳实践

Python FastAPI 入门教程：异步数据库操作最佳实践

FastAPI 教程

本教程针对有Python基础的初学者，无Web开发经验，详细讲解如何使用FastAPI进行异步数据库操作。通过生活化类比、动手实践和简单示例，帮助快速上手并建立信心。

FastAPI 入门教程：异步数据库操作最佳实践

欢迎来到FastAPI世界！

如果你已经掌握了Python基础语法，但从未接触过Web开发，那么你来对地方了。FastAPI是一个现代、快速的Web框架，特别适合初学者入门。本教程将带你从零开始，一步步构建一个异步数据库操作的应用，让你在短时间内体验成功的喜悦。

第1部分：什么是FastAPI？

生活化类比：想象FastAPI是你的个人助手，它帮你快速搭建一个“餐厅点餐系统”。你只需要告诉它菜单（API路由）和菜品做法（代码），它就能自动生成菜单册（文档），并高效处理顾客订单（请求）。

关键概念：

API：就像餐厅的菜单，定义了可以点什么的规则。
异步：在厨房做菜时，你可以同时准备多个菜品，而不是等一个做完再下一个，这样可以服务更多顾客。

为什么适合初学者：FastAPI自动生成文档、类型提示让代码更易读，且支持异步编程，提升效率。

第2部分：安装和设置

在开始之前，确保你有Python 3.7或更高版本。打开终端（或命令提示符），输入以下命令来安装FastAPI和相关工具。

pip install fastapi uvicorn sqlalchemy databases aiosqlite

fastapi：核心框架。
uvicorn：服务器，用来运行你的应用。
sqlalchemy 和 databases：用于数据库操作。
aiosqlite：异步SQLite数据库驱动，适合本地测试。

成功时刻：运行命令后，如果没有错误，你就成功安装了所有依赖！现在可以开始写代码了。

第3部分：创建第一个FastAPI应用

让我们快速上手，写一个简单的“Hello World”API。创建一个新文件 main.py。

from fastapi import FastAPI

app = FastAPI()  # 创建FastAPI应用实例

@app.get("/")  # 定义一个GET请求的路由，路径为根目录
async def read_root():
    return {"message": "Hello, World!"}

解释：

@app.get("/")：这就像一个指示牌，告诉FastAPI当有人访问网站首页时，调用下面的函数。
async def：这是一个异步函数，意味着它可以在等待时处理其他任务。

运行应用：在终端中，导航到 main.py 所在目录，运行：

uvicorn main:app --reload

--reload：启用热重载，这样你修改代码后服务器会自动重启。

成功时刻：打开浏览器访问 http://127.0.0.1:8000，你会看到 {"message": "Hello, World!"} 显示出来！恭喜，你的第一个API已上线。

第4部分：异步数据库操作基础

生活化类比：数据库就像一个图书馆，而异步操作就像你借书时，不用在柜台傻等管理员找书，而是可以先去浏览其他书架，等书找到了再通知你。

什么是异步数据库操作？ 在FastAPI中，你可以用 async 和 await 来非阻塞地执行数据库查询，让服务器能同时处理多个请求，提高性能。

关键步骤：

设置数据库：我们使用SQLite作为本地数据库，因为它简单易用。
定义模型：描述数据的结构，就像图书馆的图书目录。
连接数据库：建立与数据库的通信。
执行查询：异步地读取或写入数据。

第5部分：异步数据库操作示例

让我们扩展之前的应用，添加一个简单的数据库来存储用户信息。更新 main.py。

from fastapi import FastAPI
from databases import Database
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String
import asyncio

# 初始化FastAPI应用
app = FastAPI()

# 数据库URL，使用SQLite数据库
DATABASE_URL = "sqlite:///./test.db"

# 创建数据库实例
database = Database(DATABASE_URL)

# 定义用户表结构
metadata = MetaData()
users = Table(
    "users",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String),
    Column("email", String),
)

# 启动应用时连接数据库
@app.on_event("startup")
async def startup():
    await database.connect()
    # 创建表（如果不存在）
    engine = create_engine(DATABASE_URL)
    metadata.create_all(engine)
    print("数据库已连接并表已创建")

# 关闭应用时断开数据库连接
@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()
    print("数据库已断开")

# 路由：添加用户
@app.post("/users/")
async def create_user(name: str, email: str):
    query = users.insert().values(name=name, email=email)
    user_id = await database.execute(query)  # 异步执行插入操作
    return {"id": user_id, "name": name, "email": email}

# 路由：获取所有用户
@app.get("/users/")
async def read_users():
    query = users.select()
    results = await database.fetch_all(query)  # 异步执行查询操作
    return {"users": results}

# 保留之前的Hello World路由
@app.get("/")
async def read_root():
    return {"message": "Hello, World! 试试访问 /users/ 来管理用户吧！"}

解释：

我们定义了一个 users 表来存储用户数据。
@app.on_event("startup") 和 @app.on_event("shutdown")：在应用启动和关闭时自动连接和断开数据库。
await database.execute(query)：异步执行数据库插入操作。
await database.fetch_all(query)：异步获取所有用户数据。

运行并测试：重启服务器（如果已在运行，它会自动重载）。

访问 http://127.0.0.1:8000，你会看到新消息。
使用工具如Postman或浏览器扩展，发送POST请求到 http://127.0.0.1:8000/users/，参数 name 和 email（例如：name=John&email=john@example.com）。
然后访问 http://127.0.0.1:8000/users/ 查看所有用户。

成功时刻：当你成功添加用户并查看到列表时，你就完成了异步数据库操作！这展示了FastAPI的高效和易用性。

第6部分：最佳实践

在异步数据库操作中，遵循以下最佳实践能确保应用稳定和高效：

使用连接池：想象数据库连接就像咖啡店的员工，连接池让一群员工随时待命，而不是每次点单都招聘新人。FastAPI的 databases 库默认管理连接池。

错误处理：添加异常处理，避免应用崩溃。例如：

from fastapi import HTTPException

@app.post("/users/")
async def create_user(name: str, email: str):
    try:
        query = users.insert().values(name=name, email=email)
        user_id = await database.execute(query)
        return {"id": user_id, "name": name, "email": email}
    except Exception as e:
        raise HTTPException(status_code=500, detail="添加用户失败")

避免阻塞操作：在异步函数中，避免长时间运行的同步代码，否则会阻塞整个应用。如果需要，使用 asyncio.to_thread 或专用线程。
定期清理：在关闭时断开数据库连接，释放资源。
测试异步代码：使用 pytest-asyncio 等工具进行异步测试。

生活化类比：这就像管理一个团队：计划好任务（连接池），准备好应对问题（错误处理），并确保工作不中断（避免阻塞）。

第7部分：总结和下一步

恭喜你！通过本教程，你学会了：

安装和运行FastAPI应用。
创建简单API路由。
设置异步数据库操作，从SQLite中读写数据。
实践了最佳实践，如连接池和错误处理。

建立信心：记住，每个成功运行的示例都是你学习路上的里程碑。FastAPI让Web开发变得直观有趣。

下一步建议：

探索FastAPI的更多功能，如请求验证、身份认证。
尝试使用其他数据库，如PostgreSQL，并学习更复杂的查询。
阅读官方文档和社区教程，加深理解。

最后的成功时刻：现在，你可以自豪地说：“我用FastAPI构建了一个带异步数据库的Web应用！” 继续实践，享受编程的乐趣吧！

如果有任何问题，欢迎随时回看教程步骤。Happy coding！

上一章 13.3 异步依赖项与上下文管理器

下一章 13.5 并发与并行处理