FastAPI 快速开发指南：MySQL 数据库集成实践

在现代化 Web 开发中，FastAPI 以其高性能、易用性和自动生成 API 文档的特性，成为构建 Web API 的热门选择。而 MySQL 作为成熟的开源关系型数据库，凭借其稳定性、事务支持和广泛的社区生态，成为后端数据存储的首选方案。本文将系统阐述如何在 FastAPI 项目中高效集成 MySQL 数据库，从基础配置到高级优化，为开发者提供完整的实践指南。

一、环境准备与依赖管理

1.1 项目初始化

FastAPI 项目通常基于 Python 3.7+ 环境，推荐使用虚拟环境隔离依赖。通过以下命令创建并激活虚拟环境：

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

1.2 依赖安装

核心依赖包括 FastAPI、Uvicorn（ASGI 服务器）和 MySQL 连接驱动。推荐使用 pymysql 或 mysql-connector-python：

pip install fastapi uvicorn pymysql sqlalchemy  # SQLAlchemy 用于 ORM

若需异步支持，可安装 asyncpg 或 aiomysql，但需注意 FastAPI 默认同步模型下直接使用 pymysql 更简单。

1.3 数据库配置

在项目根目录创建 .env 文件存储敏感信息（需添加到 .gitignore）：

DB_HOST=localhost
DB_PORT=3306
DB_USER=root
DB_PASSWORD=your_password
DB_NAME=fastapi_db

通过 python-dotenv 加载配置：

from dotenv import load_dotenv
import os
load_dotenv()
DB_CONFIG = {
    "host": os.getenv("DB_HOST"),
    "port": int(os.getenv("DB_PORT")),
    "user": os.getenv("DB_USER"),
    "password": os.getenv("DB_PASSWORD"),
    "database": os.getenv("DB_NAME")
}

二、同步模式下的 MySQL 连接

2.1 直接连接（简单场景）

对于轻量级应用，可直接使用 pymysql 建立连接：

import pymysql
from fastapi import FastAPI
app = FastAPI()
def get_db_connection():
    return pymysql.connect(
        host=DB_CONFIG["host"],
        port=DB_CONFIG["port"],
        user=DB_CONFIG["user"],
        password=DB_CONFIG["password"],
        database=DB_CONFIG["database"],
        cursorclass=pymysql.cursors.DictCursor
    )
@app.get("/users")
def get_users():
    conn = get_db_connection()
    try:
        with conn.cursor() as cursor:
            cursor.execute("SELECT * FROM users")
            return cursor.fetchall()
    finally:
        conn.close()

缺点：每次请求创建新连接，性能低下。

2.2 连接池优化

使用 DBUtils 或 SQLAlchemy 的连接池提升性能：

from dbutils.pooled_db import PooledDB
pool = PooledDB(
    creator=pymysql,
    maxconnections=5,
    mincached=2,
    host=DB_CONFIG["host"],
    user=DB_CONFIG["user"],
    password=DB_CONFIG["password"],
    database=DB_CONFIG["database"]
)
@app.get("/users")
def get_users():
    conn = pool.connection()
    try:
        with conn.cursor() as cursor:
            cursor.execute("SELECT * FROM users")
            return cursor.fetchall()
    finally:
        conn.close()  # 实际归还到连接池

三、异步模式下的 MySQL 集成

3.1 使用 aiomysql

FastAPI 原生支持异步，搭配 aiomysql 可实现非阻塞 I/O：

import aiomysql
from fastapi import FastAPI, Depends
from contextlib import asynccontextmanager
app = FastAPI()
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 初始化连接池
    pool = await aiomysql.create_pool(
        host=DB_CONFIG["host"],
        port=DB_CONFIG["port"],
        user=DB_CONFIG["user"],
        password=DB_CONFIG["password"],
        db=DB_CONFIG["database"],
        minsize=5,
        maxsize=10
    )
    async with pool:
        app.state.pool = pool
        yield
        pool.close()
        await pool.wait_closed()
app = FastAPI(lifespan=lifespan)
async def get_db():
    return app.state.pool.acquire()
@app.get("/users")
async def get_users(db=Depends(get_db)):
    async with db as conn:
        async with conn.cursor() as cursor:
            await cursor.execute("SELECT * FROM users")
            return await cursor.fetchall()

3.2 使用 SQLAlchemy 2.0 异步 API

SQLAlchemy 2.0+ 原生支持异步：

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from fastapi import FastAPI, Depends
DATABASE_URL = f"mysql+aiomysql://{DB_CONFIG['user']}:{DB_CONFIG['password']}@{DB_CONFIG['host']}:{DB_CONFIG['port']}/{DB_CONFIG['database']}"
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
app = FastAPI()
@app.get("/users")
async def get_users(db: AsyncSession = Depends(get_db)):
    result = await db.execute("SELECT * FROM users")
    return result.scalars().all()

四、安全与最佳实践

4.1 防止 SQL 注入

始终使用参数化查询：

# 错误方式（易受注入攻击）
query = f"SELECT * FROM users WHERE username = '{username}'"
# 正确方式
cursor.execute("SELECT * FROM users WHERE username = %s", (username,))

4.2 敏感数据保护

• 加密存储密码（使用 bcrypt 或 passlib）
• 限制数据库用户权限（仅授予必要权限）
• 定期轮换数据库凭证

4.3 性能监控

• 启用 MySQL 慢查询日志
• 使用 Prometheus + Grafana 监控 API 响应时间
• 对高频查询添加缓存（如 Redis）

五、完整项目示例

5.1 项目结构

fastapi_mysql/
├── main.py          # 入口文件
├── models.py        # 数据模型
├── schemas.py       # Pydantic 模型
├── database.py      # 数据库连接
├── .env             # 环境变量
└── requirements.txt

5.2 核心代码

database.py:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from .config import settings
SQLALCHEMY_DATABASE_URL = f"mysql+pymysql://{settings.DB_USER}:{settings.DB_PASSWORD}@{settings.DB_HOST}:{settings.DB_PORT}/{settings.DB_NAME}"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

main.py:

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from .database import get_db
from .models import User
from .schemas import UserCreate
app = FastAPI()
@app.post("/users/")
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = User(**user.dict())
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user
@app.get("/users/{user_id}")
def read_user(user_id: int, db: Session = Depends(get_db)):
    return db.query(User).filter(User.id == user_id).first()

六、部署建议

1. 容器化部署：使用 Docker 封装应用与 MySQL 服务

   FROM python:3.9-slim
   WORKDIR /app
   COPY . .
   RUN pip install -r requirements.txt
   CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

2. 生产环境优化：

   • 启用 Gzip 压缩
   • 配置 HTTPS
   • 使用连接池大小适配并发量

3. 扩展方案：

   • 读写分离（主从架构）
   • 分库分表（如使用 Vitess）

总结

FastAPI 与 MySQL 的集成可灵活适配从简单 CRUD 到高并发场景的需求。开发者应根据项目规模选择同步/异步模式，合理配置连接池，并始终遵循安全最佳实践。通过 SQLAlchemy 等 ORM 工具，可进一步提升开发效率与代码可维护性。实际项目中，建议结合 CI/CD 流水线实现自动化测试与部署，确保系统稳定性。