FastAPI 结合 MySQL：高效构建 Web API 的完整指南

在当今快速迭代的软件开发环境中，构建高效、可扩展的 Web API 成为开发者的重要技能。FastAPI，作为一款基于 Python 的现代 Web 框架，以其高性能、易用性和自动生成 API 文档的特性，迅速成为开发 Web API 的首选工具。而 MySQL，作为一款广泛使用的开源关系型数据库，以其稳定性、可靠性和丰富的功能集，为 Web 应用提供了强大的数据存储支持。本文将详细介绍如何使用 FastAPI 快速开发 Web API 项目，并连接 MySQL 数据库，帮助开发者高效构建数据驱动的 Web 应用。

一、环境准备与项目初始化

1. 环境准备

在开始之前，确保你的开发环境已安装 Python 3.7+ 和 pip 包管理工具。此外，建议使用虚拟环境来管理项目依赖，避免不同项目间的依赖冲突。可以通过以下命令创建并激活虚拟环境：

python -m venv venv
source venv/bin/activate  # Linux/macOS
# 或
venv\Scripts\activate  # Windows

2. 安装 FastAPI 和 Uvicorn

FastAPI 依赖于 Uvicorn 作为 ASGI 服务器来运行应用。通过 pip 安装 FastAPI 和 Uvicorn：

pip install fastapi uvicorn

3. 安装 MySQL 客户端库

为了连接 MySQL 数据库，我们需要安装一个 MySQL 客户端库，如 pymysql 或 mysql-connector-python。这里以 pymysql 为例：

pip install pymysql

4. 初始化 FastAPI 项目

创建一个新的目录作为项目根目录，并在其中创建 main.py 文件，作为应用的入口点。在 main.py 中，导入 FastAPI 并创建一个 FastAPI 实例：

from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def read_root():
    return {"Hello": "World"}

运行应用：

uvicorn main:app --reload

访问 http://127.0.0.1:8000，你应该能看到 {"Hello": "World"} 的响应。

二、连接 MySQL 数据库

1. 配置数据库连接

在项目中创建一个 database.py 文件，用于配置和管理数据库连接。使用 pymysql 创建连接池或直接建立连接：

import pymysql
from pymysql import cursors
from pymysql.connections import Connection
# 数据库配置
DB_CONFIG = {
    'host': 'localhost',
    'user': 'your_username',
    'password': 'your_password',
    'database': 'your_database',
    'charset': 'utf8mb4',
    'cursorclass': cursors.DictCursor,
}
def get_db_connection() -> Connection:
    """获取数据库连接"""
    return pymysql.connect(**DB_CONFIG)

2. 封装数据库操作

为了简化数据库操作，可以创建一个 db_utils.py 文件，封装常用的 CRUD 操作。例如，封装一个执行查询并返回结果的函数：

from typing import Any, List, Dict, Optional
from .database import get_db_connection
def execute_query(query: str, params: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]:
    """执行查询并返回结果"""
    conn = get_db_connection()
    try:
        with conn.cursor() as cursor:
            cursor.execute(query, params or {})
            result = cursor.fetchall()
        return result
    finally:
        conn.close()

三、定义数据模型与 API 路由

1. 定义数据模型

使用 Pydantic 模型来定义数据结构，这有助于数据的验证和序列化。例如，定义一个 User 模型：

from pydantic import BaseModel
class User(BaseModel):
    id: int
    name: str
    email: str

2. 创建 API 路由

在 main.py 中，定义处理用户数据的 API 路由。例如，创建一个获取所有用户的路由：

from fastapi import FastAPI, HTTPException
from typing import List
from .db_utils import execute_query
from .models import User  # 假设 User 模型在 models.py 中定义
app = FastAPI()
@app.get("/users/", response_model=List[User])
def get_users():
    """获取所有用户"""
    query = "SELECT * FROM users"
    users_data = execute_query(query)
    users = [User(**user) for user in users_data]
    return users

3. 处理异常

在数据库操作中，可能会遇到各种异常，如连接失败、查询错误等。使用 FastAPI 的 HTTPException 来处理这些异常，并向客户端返回友好的错误信息：

from fastapi import HTTPException, status
@app.get("/users/{user_id}")
def get_user(user_id: int):
    """获取单个用户"""
    query = "SELECT * FROM users WHERE id = %s"
    try:
        user_data = execute_query(query, {'id': user_id})[0]
        if not user_data:
            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
        return User(**user_data)
    except IndexError:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
    except Exception as e:
        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))

四、异步处理与性能优化

1. 异步数据库操作

FastAPI 支持异步路由，可以利用 async/await 来提高 I/O 密集型操作的性能。使用 aiomysql 库来实现异步 MySQL 操作：

pip install aiomysql

修改 database.py 以支持异步连接：

import aiomysql
from typing import AsyncGenerator
async def get_db_connection() -> AsyncGenerator[aiomysql.Connection, None]:
    """获取异步数据库连接"""
    conn = await aiomysql.connect(
        host='localhost',
        user='your_username',
        password='your_password',
        db='your_database',
        charset='utf8mb4',
        cursorclass=aiomysql.DictCursor,
    )
    try:
        yield conn
    finally:
        conn.close()

2. 异步 API 路由

定义异步 API 路由来处理异步数据库操作：

from fastapi import Depends
from aiomysql import Connection
from .database import get_db_connection
async def get_users_async(db: Connection = Depends(get_db_connection)):
    """异步获取所有用户"""
    async with db.cursor() as cursor:
        await cursor.execute("SELECT * FROM users")
        users_data = await cursor.fetchall()
    users = [User(**user) for user in users_data]
    return users
@app.get("/users/async/", response_model=List[User])
async def read_users_async(db: Connection = Depends(get_db_connection)):
    return await get_users_async(db)

五、总结与展望

通过本文的介绍，我们了解了如何使用 FastAPI 快速开发 Web API 项目，并连接 MySQL 数据库。从环境准备、数据库连接、数据模型定义到 API 路由创建，以及异步处理与性能优化，我们覆盖了开发 Web API 的关键步骤。FastAPI 的高性能和易用性，结合 MySQL 的稳定性和可靠性，为开发者提供了强大的工具集，帮助快速构建数据驱动的 Web 应用。

未来，随着技术的不断进步，我们可以期待 FastAPI 和 MySQL 在更多场景下的应用，如微服务架构、实时数据处理等。同时，开发者也应持续关注最佳实践和安全规范，确保应用的稳定性和安全性。希望本文能为你的 FastAPI 和 MySQL 开发之旅提供有益的指导。