FastAPI 实战：从零构建 Web API 并高效连接 MySQL 数据库

FastAPI 作为一款现代、快速（高性能）的 Web 框架，基于 Python 3.7+ 的类型提示功能，能够快速构建高性能的 API 服务。其核心优势包括自动生成交互式 API 文档、内置数据验证、异步支持以及与异步数据库驱动的无缝集成。对于需要连接 MySQL 数据库的 Web API 项目，FastAPI 提供了简洁高效的实现方式。本文将详细介绍如何使用 FastAPI 快速开发 Web API 项目，并重点演示如何连接 MySQL 数据库。

一、环境准备与依赖安装

1. 创建 Python 虚拟环境

为了隔离项目依赖，建议使用 Python 的虚拟环境工具（如 venv 或 conda）创建独立的开发环境。

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

2. 安装 FastAPI 与 ASGI 服务器

FastAPI 本身是一个框架，需要 ASGI 服务器（如 Uvicorn）来运行。

pip install fastapi uvicorn

3. 安装 MySQL 数据库驱动

FastAPI 支持异步操作，推荐使用 asyncmy（纯 Python 实现的异步 MySQL 驱动）或 aiomysql。这里以 asyncmy 为例：

pip install asyncmy

4. 安装 SQLAlchemy（可选）

对于复杂的数据库操作，可以使用 SQLAlchemy 作为 ORM 工具，它提供了更高级的抽象和更灵活的查询方式。

pip install sqlalchemy

二、项目结构规划

一个良好的项目结构有助于代码的可维护性和扩展性。建议采用以下结构：

fastapi_mysql_project/
├── app/
│   ├── __init__.py
│   ├── main.py          # 入口文件
│   ├── models.py        # 数据库模型定义
│   ├── schemas.py       # Pydantic 模型（数据验证）
│   ├── crud.py          # 数据库操作函数
│   └── database.py      # 数据库连接与会话管理
├── requirements.txt
└── README.md

三、连接 MySQL 数据库

1. 配置数据库连接

在 database.py 中配置 MySQL 数据库连接。这里使用异步方式连接，并利用上下文管理器管理会话。

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
DATABASE_URL = "mysql+asyncmy://username:password@localhost:3306/dbname"
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

2. 定义数据库模型

在 models.py 中使用 SQLAlchemy 定义数据库表模型。例如，定义一个简单的 User 模型：

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    email = Column(String, unique=True, index=True)

3. 创建数据库表

在项目启动时，可以使用以下代码创建数据库表（确保数据库已存在）：

from app.models import Base
from app.database import engine
async def init_db():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

可以在 main.py 中调用此函数，或在项目启动脚本中调用。

四、实现 CRUD 操作

1. 定义 Pydantic 模型

在 schemas.py 中定义用于数据验证的 Pydantic 模型。例如：

from pydantic import BaseModel
class UserBase(BaseModel):
    name: str
    email: str
class UserCreate(UserBase):
    pass
class User(UserBase):
    id: int
    class Config:
        orm_mode = True

2. 实现 CRUD 函数

在 crud.py 中实现数据库的增删改查操作。例如，实现创建用户和获取用户的函数：

from sqlalchemy.future import select
from app.models import User
from app.schemas import UserCreate
async def create_user(db, user: UserCreate):
    db_user = User(name=user.name, email=user.email)
    db.add(db_user)
    await db.commit()
    await db.refresh(db_user)
    return db_user
async def get_user_by_email(db, email: str):
    result = await db.execute(select(User).where(User.email == email))
    return result.scalar_one_or_none()

五、构建 Web API

1. 定义路由与端点

在 main.py 中定义 FastAPI 路由和端点。例如，实现创建用户和获取用户的 API：

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from app import crud, schemas
from app.database import get_db
from app.models import init_db  # 假设已导入并修改为异步初始化
app = FastAPI()
@app.on_event("startup")
async def startup_event():
    await init_db()
@app.post("/users/", response_model=schemas.User)
async def create_user(user: schemas.UserCreate, db: AsyncSession = Depends(get_db)):
    db_user = await crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return await crud.create_user(db, user=user)
@app.get("/users/{email}", response_model=schemas.User)
async def read_user(email: str, db: AsyncSession = Depends(get_db)):
    db_user = await crud.get_user_by_email(db, email=email)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

2. 运行应用

使用 Uvicorn 运行 FastAPI 应用：

uvicorn app.main:app --reload

应用启动后，可以通过 http://127.0.0.1:8000/docs 访问自动生成的交互式 API 文档。

六、最佳实践与优化建议

1. 异步操作的重要性

FastAPI 支持异步操作，能够显著提高 I/O 密集型任务的性能。在连接数据库时，务必使用异步驱动（如 asyncmy）和异步 SQLAlchemy，以避免阻塞事件循环。

2. 依赖注入与会话管理

利用 FastAPI 的依赖注入系统管理数据库会话，可以确保会话的正确关闭和资源的有效利用。避免在全局范围内保持会话打开。

3. 错误处理与日志记录

实现完善的错误处理机制，捕获并处理数据库操作中的异常。同时，配置日志记录以监控应用运行状态。

4. 安全性考虑

对输入数据进行严格验证，防止 SQL 注入等安全漏洞。使用 HTTPS 协议保护数据传输安全。

5. 性能优化

对于高频访问的 API，考虑实现缓存机制（如 Redis）以减少数据库查询次数。同时，优化数据库查询语句，避免 N+1 查询问题。

七、总结与展望

本文详细介绍了如何使用 FastAPI 快速开发 Web API 项目，并重点演示了如何连接 MySQL 数据库。通过合理的项目结构规划、异步数据库操作、CRUD 函数实现以及 Web API 构建，开发者可以高效地开发出高性能的 Web 服务。未来，随着 FastAPI 和异步 Python 生态的不断发展，我们可以期待更多高效、易用的工具和库的出现，进一步简化 Web API 的开发过程。