从零搭建FastAPI+MySQL项目：高效开发Web API全流程指南

一、FastAPI与MySQL的协同优势

FastAPI作为现代Python Web框架，以其高性能（基于Starlette与Pydantic）、自动生成OpenAPI文档和异步支持等特性，成为开发RESTful API的首选。而MySQL作为成熟的开源关系型数据库，具备高可靠性、事务支持和丰富的SQL功能。将两者结合，可构建出既高效又稳定的Web服务：

• 性能优化：FastAPI的异步特性（如async/await）与MySQL的连接池技术结合，可显著提升并发处理能力。
• 开发效率：Pydantic模型自动验证请求/响应数据，减少手动校验代码；SQLAlchemy Core或ORM提供类型安全的数据库操作。
• 生态兼容：FastAPI原生支持ASGI服务器（如Uvicorn），与MySQL的异步驱动（如aiomysql）无缝协作。

二、环境准备与依赖安装

1. 项目初始化

mkdir fastapi-mysql-demo && cd fastapi-mysql-demo
python -m venv venv
source venv/bin/activate  # Linux/Mac
# 或 venv\Scripts\activate (Windows)
pip install fastapi uvicorn sqlalchemy pymysql python-dotenv

• 关键依赖：
  • fastapi：核心框架
  • uvicorn：ASGI服务器
  • sqlalchemy：数据库工具包（支持Core与ORM）
  • pymysql：MySQL纯Python驱动（或aiomysql用于异步）
  • python-dotenv：环境变量管理

2. 配置环境变量

创建.env文件：

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

三、数据库连接与模型定义

1. 使用SQLAlchemy Core（推荐高性能场景）

# database.py
from sqlalchemy import create_engine, text
from sqlalchemy.engine import URL
from dotenv import load_dotenv
import os
load_dotenv()
# 同步连接（适用于简单项目）
def get_db_sync():
    url = URL.create(
        drivername="mysql+pymysql",
        username=os.getenv("DB_USER"),
        password=os.getenv("DB_PASSWORD"),
        host=os.getenv("DB_HOST"),
        port=os.getenv("DB_PORT"),
        database=os.getenv("DB_NAME")
    )
    engine = create_engine(url)
    return engine
# 异步连接（需SQLAlchemy 1.4+和aiomysql）
async def get_db_async():
    from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
    from sqlalchemy.orm import sessionmaker
    url = URL.create(
        drivername="mysql+aiomysql",
        # 其他参数同上
    )
    engine = create_async_engine(url, echo=True)
    AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
    return AsyncSessionLocal

2. 使用SQLAlchemy ORM（适合复杂模型）

# models.py
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime
Base = declarative_base()
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True, index=True)
    email = Column(String(100), unique=True)
    created_at = Column(DateTime, default=datetime.utcnow)

四、实现CRUD操作

1. 同步模式示例

# crud_sync.py
from sqlalchemy.orm import Session
from .models import User
def create_user(db: Session, user_data: dict):
    db_user = User(**user_data)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user
def get_user_by_id(db: Session, user_id: int):
    return db.query(User).filter(User.id == user_id).first()

2. 异步模式示例

# crud_async.py
from sqlalchemy.ext.asyncio import AsyncSession
from .models import User
async def create_user_async(db: AsyncSession, user_data: dict):
    db_user = User(**user_data)
    db.add(db_user)
    await db.commit()
    await db.refresh(db_user)
    return db_user
async def get_user_by_id_async(db: AsyncSession, user_id: int):
    result = await db.execute(
        select(User).where(User.id == user_id)
    )
    return result.scalar_one_or_none()

五、构建FastAPI路由

1. 基础路由实现

# main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from .database import get_db_sync, SessionLocal
from .crud_sync import create_user, get_user_by_id
from .schemas import UserCreate, User  # Pydantic模型
app = FastAPI()
# 依赖注入
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
@app.post("/users/", response_model=User)
def create_user_endpoint(user: UserCreate, db: Session = Depends(get_db)):
    db_user = get_user_by_id(db, user.id)  # 假设UserCreate包含id
    if db_user:
        raise HTTPException(status_code=400, detail="User already exists")
    return create_user(db, user.dict())
@app.get("/users/{user_id}", response_model=User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = get_user_by_id(db, user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

2. 异步路由示例

# main_async.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from .database import get_db_async
from .crud_async import create_user_async, get_user_by_id_async
from .schemas import UserCreate, User
app = FastAPI()
async def get_db():
    async with get_db_async() as db:
        yield db
@app.post("/users/", response_model=User)
async def create_user_endpoint(user: UserCreate, db: AsyncSession = Depends(get_db)):
    db_user = await get_user_by_id_async(db, user.id)
    if db_user:
        raise HTTPException(status_code=400, detail="User already exists")
    return await create_user_async(db, user.dict())

六、高级实践与优化

1. 连接池配置

# 同步连接池
engine = create_engine(
    url,
    pool_size=5,       # 连接池大小
    max_overflow=10,   # 超出pool_size后的最大连接数
    pool_timeout=30,   # 获取连接的超时时间（秒）
    pool_recycle=3600  # 连接回收时间（秒）
)
# 异步连接池（SQLAlchemy 1.4+）
async_engine = create_async_engine(
    url,
    pool_size=5,
    max_overflow=10,
    pool_pre_ping=True  # 每次获取连接前执行PING
)

2. 事务管理

# 同步事务示例
def update_user_email(db: Session, user_id: int, new_email: str):
    try:
        user = db.query(User).filter(User.id == user_id).with_for_update().first()
        user.email = new_email
        db.commit()
    except Exception as e:
        db.rollback()
        raise e
# 异步事务示例
async def update_user_email_async(db: AsyncSession, user_id: int, new_email: str):
    async with db.begin():
        await db.execute(
            update(User).where(User.id == user_id).values(email=new_email)
        )

3. 性能监控

• 慢查询日志：在MySQL配置中启用slow_query_log
• FastAPI中间件：记录请求处理时间
  ```python
  from fastapi import Request
  from fastapi.middleware import Middleware
  from fastapi.middleware.base import BaseHTTPMiddleware
  import time

class TimingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
response.headers[“X-Process-Time”] = str(process_time)
return response

app.add_middleware(TimingMiddleware)

## 七、部署建议
1. **生产环境配置**：
   - 使用Gunicorn + Uvicorn工人模式：
     ```bash
     gunicorn -k uvicorn.workers.UvicornWorker -w 4 -b :8000 main:app

• 配置Nginx作为反向代理

1. 数据库优化：

   • 主从复制提升读取性能
   • 分库分表策略应对大数据量

2. 安全实践：

   • 启用SSL/TLS加密
   • 使用pydantic的SecretStr处理敏感字段
   • 限制API访问频率（如slowapi库）

八、常见问题解决方案

1. 连接失败排查：

   • 检查MySQL服务是否运行：systemctl status mysql
   • 验证用户权限：GRANT ALL PRIVILEGES ON fastapi_demo.* TO 'root'@'%';
   • 防火墙设置：确保3306端口开放

2. 异步驱动兼容性：

   • SQLAlchemy 1.4+才支持原生异步
   • aiomysql需与asyncmy或pymysql的异步版本配合

3. 性能瓶颈定位：

   • 使用cProfile分析代码热点
   • MySQL的EXPLAIN分析慢查询

通过以上步骤，开发者可快速构建一个基于FastAPI和MySQL的高性能Web API项目。实际开发中，建议结合单元测试（如pytest）和持续集成（CI）流程，确保代码质量与稳定性。