FastAPI 结合 MySQL：高效构建 Web API 的全流程指南

在当今快速迭代的软件开发环境中，FastAPI 凭借其高性能、易用性和强大的类型提示支持，成为了构建 Web API 的热门选择。而 MySQL，作为一款广泛使用的开源关系型数据库，以其稳定性、可靠性和丰富的功能集，为数据存储提供了坚实的后盾。本文将详细介绍如何使用 FastAPI 快速开发一个 Web API 项目，并重点讲解如何连接 MySQL 数据库，实现数据的增删改查（CRUD）操作。

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

1. 环境准备

在开始之前，确保你的开发环境中已安装 Python（建议版本 3.8+）和 pip（Python 包管理工具）。此外，还需要安装 MySQL 数据库服务器，或者使用云服务提供的 MySQL 实例。

2. 创建项目目录

首先，创建一个新的项目目录，并在其中初始化一个虚拟环境，以隔离项目依赖：

mkdir fastapi_mysql_project
cd fastapi_mysql_project
python -m venv venv
source venv/bin/activate  # Linux/Mac
# 或者 venv\Scripts\activate  # Windows

3. 安装 FastAPI 和 Uvicorn

FastAPI 需要配合 ASGI 服务器运行，这里我们选择 Uvicorn：

pip install fastapi uvicorn

4. 安装 MySQL 客户端库

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

pip install pymysql

二、连接 MySQL 数据库

1. 配置数据库连接

在项目的根目录下创建一个 database.py 文件，用于配置和管理数据库连接：

import pymysql
from pymysql.cursors import DictCursor
from typing import Optional
class Database:
    def __init__(self, host: str, user: str, password: str, database: str):
        self.host = host
        self.user = user
        self.password = password
        self.database = database
        self.connection: Optional[pymysql.Connection] = None
    def connect(self):
        self.connection = pymysql.connect(
            host=self.host,
            user=self.user,
            password=self.password,
            database=self.database,
            cursorclass=DictCursor
        )
    def close(self):
        if self.connection:
            self.connection.close()
    def get_connection(self):
        if not self.connection or not self.connection.open:
            self.connect()
        return self.connection

2. 初始化数据库连接

在项目的入口文件（如 main.py）中，初始化数据库连接：

from fastapi import FastAPI
from database import Database
app = FastAPI()
db = Database(host='localhost', user='your_username', password='your_password', database='your_database')
@app.on_event("startup")
async def startup_event():
    db.connect()
@app.on_event("shutdown")
async def shutdown_event():
    db.close()

三、定义数据模型与 CRUD 操作

1. 定义数据模型

使用 Pydantic 模型来定义 API 的请求和响应数据结构。例如，定义一个简单的用户模型：

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

2. 实现 CRUD 操作

在 main.py 或单独的模块中，实现针对用户表的 CRUD 操作：

from fastapi import HTTPException
from typing import List
# 假设已经有一个获取数据库连接的方法 get_db()
def get_db():
    return db.get_connection()
def get_users() -> List[User]:
    conn = get_db()
    with conn.cursor() as cursor:
        cursor.execute("SELECT * FROM users")
        users = cursor.fetchall()
        return [User(**user) for user in users]
def create_user(user: User) -> User:
    conn = get_db()
    with conn.cursor() as cursor:
        sql = "INSERT INTO users (name, email) VALUES (%s, %s)"
        cursor.execute(sql, (user.name, user.email))
        conn.commit()
        user.id = cursor.lastrowid
        return user
# 类似地实现 update_user 和 delete_user 方法

四、构建 API 路由

1. 定义 API 路由

在 main.py 中，使用 FastAPI 的装饰器来定义 API 路由：

from fastapi import APIRouter, HTTPException
router = APIRouter()
@router.get("/users", response_model=List[User])
def read_users():
    return get_users()
@router.post("/users", response_model=User)
def create_user_route(user: User):
    return create_user(user)
# 类似地定义更新和删除用户的路由
app.include_router(router)

2. 运行 API 服务器

使用 Uvicorn 运行 FastAPI 服务器：

uvicorn main:app --reload

五、项目优化与扩展建议

1. 使用连接池

对于高并发的应用，考虑使用连接池来管理数据库连接，以提高性能和资源利用率。

2. 异步支持

FastAPI 原生支持异步操作，可以考虑使用 async 和 await 关键字来异步执行数据库操作，进一步提升性能。

3. 安全性考虑

• 使用 HTTPS 协议保护数据传输。
• 实现身份验证和授权机制，如 JWT。
• 对输入数据进行验证和清理，防止 SQL 注入等攻击。

4. 日志记录与监控

• 实现详细的日志记录，便于问题追踪和性能分析。
• 使用监控工具（如 Prometheus、Grafana）来监控 API 的性能和健康状况。

5. 文档与测试

• 使用 FastAPI 自动生成的 API 文档（Swagger UI）来方便地测试和文档化 API。
• 编写单元测试和集成测试，确保代码的质量和稳定性。

六、总结

通过本文的介绍，你已经了解了如何使用 FastAPI 快速开发一个 Web API 项目，并成功连接 MySQL 数据库实现数据的 CRUD 操作。FastAPI 的高性能和易用性，结合 MySQL 的稳定性和可靠性，为构建现代 Web 应用提供了强大的支持。希望本文能对你的开发工作有所帮助，并激发你探索更多高级特性的兴趣。