摘要：本文详细介绍了使用 FastAPI 框架连接 Docker 容器中的 MySQL 数据库的完整实践流程。内容涵盖数据库连接配置、SQLAlchemy ORM 模型定义、依赖注入与会话管理、查询与插入操作实现，以及常见错误解决方法。通过具体的代码示例和步骤说明，帮助初学者快速掌握 FastAPI 与 MySQL 的集成开发，实现完整的 RESTful API 接口。
📌 目录
- 一、数据库连接配置
- 二、ORM 模型与自动建表
- 三、依赖注入与会话管理
- 四、查询操作（Query）
- 五、插入操作（Add + Commit）
- 六、关于 async def 与 def 的选择
- 七、常见错误及解决方法
- 八、成果总结结
一、数据库连接配置
1.1 连接字符串（Database URL）
SQLALCHEMY_DATABASE_URL = (
"mysql+pymysql://root:你的密码@localhost:3306/fastapi_demo?charset=utf8mb4"
)
| mysql+pymysql | 使用 pymysql 驱动（同步） |
| root:你的密码 | 用户名和密码（与 Docker 容器一致） |
| localhost:3306 | 数据库地址和端口（容器端口映射到宿主机） |
| fastapi_demo | 数据库名称（需提前 CREATE DATABASE） |
| charset=utf8mb4 | 字符集，支持 emoji 等特殊字符 |
1.2 引擎（Engine）与会话工厂（SessionLocal）
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
| engine | 数据库连接池，管理物理连接 |
| SessionLocal | 工厂函数，每次调用创建一个新的数据库会话（工作区） |
| autocommit=False | 手动控制事务（提交/回滚） |
| autoflush=False | 禁止自动刷新，减少不必要的 SQL |
二、ORM 模型与自动建表
2.1 声明基类
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
- Base 是所有模型的基类，内部维护 metadata 注册表，记录所有子类。
2.2 定义模型
from sqlalchemy import Column, Integer, String
class Book(Base):
__tablename__ = "books"
id = Column(Integer, primary_key=True, index=True)
title = Column(String(100), index=True)
author = Column(String(50), index=True)
| id | Integer | 主键，自增，带索引 |
| title | String(100) | 书名，最长100字符，带索引 |
| author | String(50) | 作者，最长50字符，带索引 |
2.3 自动创建表
Base.metadata.create_all(bind=engine)
- 扫描所有模型，为未存在的表执行 CREATE TABLE。
- 幂等操作，不会删除或修改已有数据。。
三、依赖注入与会话管理
3.1 使用 yield 管理会话生命周期
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
- 作用：每次请求创建一个新会话，请求结束后自动关闭。
- yield 原理：函数在 yield db 处暂停，将 db 交给路由函数；路由执行完后回到 finally 关闭会话。
3.2 在路由中使用 Depends
from fastapi import Depends
@app.get("/books/db")
def get_books_from_db(db: Session = Depends(get_db)):
books = db.query(Book).all()
return books
- Depends(get_db) 触发依赖注入，FastAPI 自动调用 get_db() 并将返回值注入到 db 参数。
- 开发者无需手动管理连接生命周期。
四、查询操作（Query）
books = db.query(Book).all()
| db.query(Book) | 构建查询对象（未执行 SQL） |
| .all() | 执行查询，返回所有记录列表 |
| .first() | 返回第一条记录或 None |
| .one() | 返回唯一记录，若数目不为1则抛异常 |
| .count() | 返回记录总数 |
- 等价 SQL：SELECT * FROM books;
五、插入操作（Add + Commit）
5.1 定义 Pydantic 模型（请求体校验）
from pydantic import BaseModel
class BookCreate(BaseModel):
title: str
author: str
- 为什么不用 ORM 模型（Book）？
因为 Book 包含 id（自增主键），不应由前端传入。Pydantic可精准控制接收字段，并自动校验类型与必填。
5.2 标准 POST 接口流程
@app.post("/books/db")
def add(book: BookCreate, db: Session = Depends(get_db)):
# 1. 转换 Pydantic → ORM
new_book = Book(title=book.title, author=book.author)
# 2. 加入会话（内存标记）
db.add(new_book)
# 3. 提交事务（执行 INSERT）
db.commit()
# 4. 刷新对象，获取自增 ID
db.refresh(new_book)
# 5. 返回所有图书（示例）
books = db.query(Book).all()
return books
| 1 | Book(title=…, author=…) | 创建 ORM 实例 |
| 2 | db.add() | 将对象加入待处理队列（未执行 SQL） |
| 3 | db.commit() | 提交事务，真正 INSERT |
| 4 | db.refresh() | 从数据库刷新，获取自动生成字段（如 id） |
| 5 | db.query().all() | 查询并返回最新列表 |
六、关于 async def 与 def 的选择
| 使用同步驱动（如 pymysql） | def | 阻塞操作会阻塞事件循环，FastAPI 将 def 放入外部线程池，不阻塞主循环 |
| 使用异步驱动（如 asyncmy + sqlalchemy.ext.asyncio） | async def + await | 真正的非阻塞 I/O，需配合异步 ORM |
| 仅返回静态数据，无 I/O | def 或 async def 均可 | 不影响性能，通常 def 更简洁 |
- 当前项目：使用 pymysql（同步），路由函数均采用 def，这是最正确且高效的做法。
七、常见错误及解决方法
| NameError: name 'get_db' is not defined | 在 get_db 定义前使用了它 | 将 def get_db() 移到所有路由函数之前 |
| BaseModel.__init__() takes 1 positional argument but 3 were given | 使用位置参数实例化 Pydantic 模型 | 改用关键字参数：BookCreate(title=…, author=…) |
| UnmappedInstanceError: Class '…BookCreate' is not mapped | 将 Pydantic 对象传给了 db.add() | db.add() 只接受 ORM 对象（Book） |
| Swagger UI 一直转圈 | 文件名含中文/特殊字符 | 将文件名改为纯英文（如 main.py） |
| 422 Validation Error 显示 null | Swagger UI 渲染问题 | 查看终端日志或用 curl 测试，获取真实错误信息 |
| Docker 镜像拉取超时/403 | 镜像加速器配置错误或代理干扰 | 配置中科大镜像加速器，或关闭代理 |
