1. 项目概述：FastAPI与MySQL的ORM实战

第一次接触FastAPI的开发者常会遇到一个关键问题：如何将应用程序的数据持久化存储？这就是ORM（对象关系映射）技术的用武之地。本教程将带你从零开始，使用SQLAlchemy作为ORM工具，实现FastAPI与MySQL数据库的无缝对接。

我最近在开发一个内部员工管理系统时，就采用了这套技术栈。当时我们需要快速搭建一个既能处理高并发请求，又能保证数据一致性的后端服务。FastAPI的异步特性加上SQLAlchemy的成熟ORM方案，完美满足了需求。

2. 环境准备与MySQL配置

2.1 安装必要依赖

在开始前，确保已安装Python 3.7+和MySQL 5.7+。然后通过pip安装所需包：

pip install fastapi sqlalchemy pymysql python-dotenv uvicorn

这里特别说明包的选择理由：

• pymysql：纯Python实现的MySQL驱动，比mysqlclient更易安装
• python-dotenv：用于管理数据库连接等敏感配置
• uvicorn：ASGI服务器，用于运行FastAPI应用

2.2 MySQL数据库设置

首先登录MySQL创建数据库和用户：

CREATE DATABASE fastapi_demo CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; CREATE USER 'fastapi_user'@'%' IDENTIFIED BY 'your_strong_password'; GRANT ALL PRIVILEGES ON fastapi_demo.* TO 'fastapi_user'@'%'; FLUSH PRIVILEGES;

注意：生产环境应该限制IP访问（将'%'改为具体IP），并使用更复杂的密码

3. SQLAlchemy核心配置

3.1 数据库连接配置

在项目根目录创建.env文件：

DB_HOST=localhost DB_PORT=3306 DB_USER=fastapi_user DB_PASSWORD=your_strong_password DB_NAME=fastapi_demo

然后创建database.py：

from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker import os from dotenv import load_dotenv load_dotenv() DB_URL = f"mysql+pymysql://{os.getenv('DB_USER')}:{os.getenv('DB_PASSWORD')}@{os.getenv('DB_HOST')}:{os.getenv('DB_PORT')}/{os.getenv('DB_NAME')}?charset=utf8mb4" engine = create_engine(DB_URL, pool_pre_ping=True, pool_recycle=3600) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base()

关键参数说明：

• pool_pre_ping=True：自动检测连接是否有效
• pool_recycle=3600：每小时重建连接，避免MySQL 8小时断开问题
• charset=utf8mb4：支持完整的Unicode字符集

3.2 模型定义最佳实践

创建models.py定义数据模型：

from sqlalchemy import Column, Integer, String from database import Base class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True, index=True) username = Column(String(50), unique=True, nullable=False) email = Column(String(100), unique=True, index=True) hashed_password = Column(String(255), nullable=False) is_active = Column(Boolean, default=True) __table_args__ = {'mysql_collate': 'utf8mb4_unicode_ci'}

设计建议：

1. 总是显式定义__tablename__
2. 字符串字段明确长度限制
3. 敏感字段如密码使用hashed_前缀
4. 添加index=True到常用查询字段
5. 为MySQL指定正确的排序规则

4. FastAPI集成实现

4.1 依赖注入配置

在main.py中设置数据库会话依赖：

from fastapi import Depends, FastAPI from sqlalchemy.orm import Session app = FastAPI() def get_db(): db = SessionLocal() try: yield db finally: db.close() @app.get("/users/{user_id}") def read_user(user_id: int, db: Session = Depends(get_db)): user = db.query(User).filter(User.id == user_id).first() if not user: raise HTTPException(status_code=404, detail="User not found") return user

4.2 CRUD操作完整示例

扩展main.py实现完整CRUD：

from fastapi import HTTPException, status from pydantic import BaseModel from typing import List # 请求响应模型 class UserCreate(BaseModel): username: str email: str password: str class UserResponse(BaseModel): id: int username: str email: str is_active: bool @app.post("/users/", response_model=UserResponse, status_code=status.HTTP_201_CREATED) def create_user(user: UserCreate, db: Session = Depends(get_db)): # 密码哈希处理 hashed_password = get_password_hash(user.password) db_user = User( username=user.username, email=user.email, hashed_password=hashed_password ) db.add(db_user) db.commit() db.refresh(db_user) return db_user @app.get("/users/", response_model=List[UserResponse]) def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): users = db.query(User).offset(skip).limit(limit).all() return users

5. 高级技巧与性能优化

5.1 异步Session支持

FastAPI天生支持异步，我们可以使用SQLAlchemy的异步功能：

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession async_engine = create_async_engine( "mysql+aiomysql://user:pass@localhost/db", echo=True ) async def get_async_db(): async with AsyncSession(async_engine) as session: yield session @app.get("/async-users/{user_id}") async def read_async_user(user_id: int, db: AsyncSession = Depends(get_async_db)): result = await db.execute(select(User).filter(User.id == user_id)) user = result.scalars().first() if not user: raise HTTPException(status_code=404) return user

5.2 连接池优化配置

对于生产环境，建议调整连接池参数：

from sqlalchemy.pool import QueuePool engine = create_engine( DB_URL, poolclass=QueuePool, pool_size=10, max_overflow=20, pool_timeout=30, pool_pre_ping=True )

参数说明：

• pool_size：保持的连接数
• max_overflow：允许超过pool_size的连接数
• pool_timeout：获取连接的超时时间(秒)

6. 常见问题排查

6.1 连接泄漏检测

在开发环境添加以下中间件检测未关闭的Session：

@app.middleware("http") async def db_session_middleware(request: Request, call_next): response = await call_next(request) if request.state.db.is_active: request.state.db.close() print("警告：未关闭的数据库会话！") return response

6.2 性能问题排查

使用SQLAlchemy的事件监听SQL执行：

from sqlalchemy import event @event.listens_for(engine, "before_cursor_execute") def before_cursor_execute(conn, cursor, statement, parameters, context, executemany): conn.info.setdefault('query_start_time', []).append(time.time()) @event.listens_for(engine, "after_cursor_execute") def after_cursor_execute(conn, cursor, statement, parameters, context, executemany): total = time.time() - conn.info['query_start_time'].pop(-1) if total > 0.5: # 超过500ms的查询 print(f"慢查询警告：{statement} 耗时{total:.2f}s")

7. 项目结构建议

成熟的FastAPI项目推荐结构：

/myapp ├── alembic/ # 数据库迁移 ├── app/ │ ├── core/ # 核心配置 │ ├── models/ # 数据模型 │ ├── schemas/ # Pydantic模型 │ ├── api/ # 路由端点 │ ├── db/ # 数据库相关 │ └── main.py # FastAPI实例 ├── tests/ # 测试代码 ├── .env # 环境变量 └── requirements.txt # 依赖列表

8. 安全注意事项

1. 永远不要在前端返回hashed_password
2. 使用HTTPS加密传输
3. 实现适当的权限控制
4. 对用户输入进行严格验证
5. 定期备份数据库

我在实际项目中遇到过因为没有及时关闭Session导致连接池耗尽的情况。后来通过添加中间件检测和设置合理的连接超时解决了这个问题。另一个教训是在模型定义时没有设置字符串长度限制，导致在MySQL 5.7上出现了截断问题。