FastAPI与MySQL实战：从零构建高性能Web API

2025年10月17日 互联网

FastAPI与MySQL实战：从零构建高性能Web API

引言：FastAPI与MySQL的黄金组合

在当今微服务架构盛行的开发环境中，FastAPI凭借其高性能、易用性和自动生成API文档的特性，迅速成为Python生态中构建Web API的首选框架。而MySQL作为最流行的开源关系型数据库，以其稳定性、扩展性和丰富的功能支持，成为后端数据存储的理想选择。将FastAPI与MySQL结合，开发者可以快速构建出高效、可靠的Web API服务。本文将详细介绍如何使用FastAPI快速开发Web API项目，并重点演示如何连接MySQL数据库，包括环境配置、依赖安装、连接池管理、CRUD操作以及错误处理等关键环节。

一、环境准备与依赖安装

1.1 创建Python虚拟环境

首先，建议为项目创建一个独立的Python虚拟环境，以避免依赖冲突。可以使用venv模块或conda来创建：

  1. # 使用venv
  2. python -m venv fastapi_mysql_env
  3. source fastapi_mysql_env/bin/activate  # Linux/Mac
  4. fastapi_mysql_env\Scripts\activate     # Windows
  6. # 或使用conda
  7. conda create --name fastapi_mysql_env python=3.9
  8. conda activate fastapi_mysql_env

1.2 安装FastAPI与Uvicorn

FastAPI需要配合ASGI服务器运行，Uvicorn是一个轻量级的ASGI服务器，适合开发和生产环境使用：

  1. pip install fastapi uvicorn

1.3 安装MySQL连接驱动

Python中连接MySQL的常用驱动有mysql-connector-python和PyMySQL。这里以PyMySQL为例：

  1. pip install pymysql

或者，如果使用SQLAlchemy作为ORM工具（推荐），则需要安装：

  1. pip install sqlalchemy pymysql

二、配置MySQL数据库连接

2.1 直接使用PyMySQL连接

如果不使用ORM，可以直接使用PyMySQL建立连接：

  1. import pymysql
  2. from pymysql import cursors
  4. # 数据库配置
  5. DB_CONFIG = {
  6.     'host': 'localhost',
  7.     'user': 'your_username',
  8.     'password': 'your_password',
  9.     'database': 'your_database',
  10.     'charset': 'utf8mb4',
  11.     'cursorclass': cursors.DictCursor
  12. }
  14. def get_db_connection():
  15.     return pymysql.connect(**DB_CONFIG)

2.2 使用SQLAlchemy配置连接池

对于生产环境，建议使用SQLAlchemy的连接池管理，以提高性能和资源利用率：

  1. from sqlalchemy import create_engine
  2. from sqlalchemy.orm import sessionmaker
  4. # 数据库URL格式：mysql+pymysql://username:password@host:port/database
  5. DATABASE_URL = "mysql+pymysql://your_username:your_password@localhost:3306/your_database"
  7. engine = create_engine(
  8.     DATABASE_URL,
  9.     pool_size=5,       # 连接池大小
  10.     max_overflow=10,   # 超出连接池大小外最多创建的连接数
  11.     pool_timeout=30,   # 等待连接的超时时间（秒）
  12.     pool_recycle=3600, # 连接回收时间（秒）
  13.     echo=True          # 打印SQL日志（开发时启用）
  14. )
  16. SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

三、FastAPI项目结构与路由设计

3.1 项目结构

一个典型的FastAPI项目结构如下：

  1. fastapi_mysql_project/
  2. ├── main.py                # 入口文件
  3. ├── models.py              # 数据模型（ORM）
  4. ├── schemas.py             # 数据验证模型（Pydantic）
  5. ├── crud.py                # 数据操作层
  6. ├── database.py            # 数据库连接配置
  7. └── requirements.txt       # 依赖文件

3.2 定义Pydantic模型

使用Pydantic定义数据验证模型，确保API输入输出的数据类型安全：

  1. from pydantic import BaseModel
  3. class ItemBase(BaseModel):
  4.     name: str
  5.     description: str | None = None
  7. class ItemCreate(ItemBase):
  8.     pass
  10. class Item(ItemBase):
  11.     id: int
  12.     owner_id: int
  14.     class Config:
  15.         orm_mode = True

3.3 定义SQLAlchemy模型

使用SQLAlchemy定义数据库表模型：

  1. from sqlalchemy import Column, Integer, String, ForeignKey
  2. from sqlalchemy.orm import relationship
  3. from .database import Base
  5. class Item(Base):
  6.     __tablename__ = "items"
  8.     id = Column(Integer, primary_key=True, index=True)
  9.     name = Column(String, index=True)
  10.     description = Column(String, index=True)
  11.     owner_id = Column(Integer, ForeignKey("users.id"))
  13.     owner = relationship("User", back_populates="items")

四、实现CRUD操作

4.1 数据操作层（CRUD）

将数据库操作封装在crud.py中，保持业务逻辑的清晰：

  1. from sqlalchemy.orm import Session
  2. from . import models, schemas
  4. def get_item(db: Session, item_id: int):
  5.     return db.query(models.Item).filter(models.Item.id == item_id).first()
  7. def get_items(db: Session, skip: int = 0, limit: int = 100):
  8.     return db.query(models.Item).offset(skip).limit(limit).all()
  10. def create_item(db: Session, item: schemas.ItemCreate, user_id: int):
  11.     db_item = models.Item(
  12.         name=item.name, 
  13.         description=item.description, 
  14.         owner_id=user_id
  15.     )
  16.     db.add(db_item)
  17.     db.commit()
  18.     db.refresh(db_item)
  19.     return db_item

4.2 定义API路由

在main.py中定义API路由，并注入数据库会话：

  1. from fastapi import Depends, FastAPI, HTTPException
  2. from sqlalchemy.orm import Session
  3. from . import crud, models, schemas
  4. from .database import SessionLocal, engine
  6. models.Base.metadata.create_all(bind=engine)
  8. app = FastAPI()
  10. def get_db():
  11.     db = SessionLocal()
  12.     try:
  13.         yield db
  14.     finally:
  15.         db.close()
  17. @app.post("/items/", response_model=schemas.Item)
  18. def create_item_for_user(
  19.     item: schemas.ItemCreate, 
  20.     user_id: int, 
  21.     db: Session = Depends(get_db)
  22. ):
  23.     db_item = crud.create_item(db, item=item, user_id=user_id)
  24.     if db_item is None:
  25.         raise HTTPException(status_code=400, detail="Error creating item")
  26.     return db_item
  28. @app.get("/items/{item_id}", response_model=schemas.Item)
  29. def read_item(item_id: int, db: Session = Depends(get_db)):
  30.     db_item = crud.get_item(db, item_id=item_id)
  31.     if db_item is None:
  32.         raise HTTPException(status_code=404, detail="Item not found")
  33.     return db_item

五、运行与测试

5.1 启动FastAPI应用

使用Uvicorn运行应用：

  1. uvicorn main:app --reload

--reload参数会在代码更改时自动重启服务器，适合开发环境。

5.2 测试API

FastAPI会自动生成交互式API文档，访问http://127.0.0.1:8000/docs即可测试API。也可以使用curl或httpie进行测试：

  1. # 创建item
  2. curl -X POST "http://127.0.0.1:8000/items/" \
  3. -H "Content-Type: application/json" \
  4. -d '{"name": "Test Item", "description": "A test item"}' \
  5. -G --data-urlencode "user_id=1"
  7. # 获取item
  8. curl "http://127.0.0.1:8000/items/1"

六、进阶优化与最佳实践

6.1 连接池管理

  • 合理设置连接池参数：根据并发量调整pool_size和max_overflow。
  • 定期回收连接：通过pool_recycle避免连接长时间闲置。

6.2 异步支持

FastAPI原生支持异步，可以使用async/await和异步数据库驱动（如asyncpg）提升性能：

  1. from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
  2. from sqlalchemy.orm import sessionmaker
  4. DATABASE_URL = "mysql+asyncpg://user:password@localhost/db"
  6. engine = create_async_engine(
  7.     DATABASE_URL,
  8.     echo=True,
  9.     future=True
  10. )
  12. AsyncSessionLocal = sessionmaker(
  13.     bind=engine,
  14.     class_=AsyncSession,
  15.     expire_on_commit=False
  16. )

6.3 错误处理与日志

  • 统一错误响应：使用FastAPI的HTTPException返回标准错误格式。
  • 日志记录：配置日志模块记录请求和数据库操作。

七、总结与展望

本文详细介绍了如何使用FastAPI快速开发Web API项目，并重点演示了与MySQL数据库的连接、配置及CRUD操作。通过合理使用SQLAlchemy的连接池管理、Pydantic的数据验证以及FastAPI的依赖注入系统，开发者可以构建出高效、可靠的Web API服务。未来，随着异步编程和Serverless架构的普及，FastAPI与MySQL的组合将在更多场景中发挥重要作用。

最新文章