FastAPI + SQLAlchemy + PyMySQL 连接 My

脚本专家

在 FastAPI 后端中接入 MySQL，不只是写一条连接串那么简单。我们需要同时解决驱动选择、ORM 映射、会话管理、事务控制以及项目结构等问题。下面以同步版为例，梳理一条从安装依赖到增删改查的完整链路。

一、依赖安装
项目需要以下核心库：

2. pip install "fastapi[standard]" sqlalchemy pymysql pydantic-settings

复制代码

如果使用新版 MySQL 认证方式（如 caching_sha2_password），可以加装 RSA 支持：

2. pip install "pymysql[rsa]"

复制代码

正式环境建议额外安装 Alembic 管理数据库迁移：

2. pip install alembic

复制代码

二、环境配置
数据库连接串和开关参数放在 .env 文件中，避免硬编码：

2. .env
3. DATABASE_URL="mysql+pymysql://root:password@127.0.0.1:3306/todo_api?charset=utf8mb4"
4. DATABASE_ECHO=false
5. DATABASE_AUTO_CREATE=false

复制代码

用 pydantic-settings 读取：

2. from pydantic_settings import BaseSettings, SettingsConfigDict
4. class Settings(BaseSettings):
5.     model_config = SettingsConfigDict(
6.         env_file=".env",
7.         env_file_encoding="utf-8",
8.     )
9.     database_url: str
10.     database_echo: bool = False
11.     database_auto_create: bool = False
13. settings = Settings()

复制代码

三、创建 Engine 与 SessionLocal
Engine 是数据库入口，管理连接池：

2. from sqlalchemy import create_engine
3. from app.core.config import settings
5. engine = create_engine(
6.     settings.database_url,
7.     echo=settings.database_echo,
8.     pool_pre_ping=True,   # 取出连接前检查是否存活
9. )

复制代码

SessionLocal 是一个会话工厂，每个请求创建一个独立会话：

2. from sqlalchemy.orm import sessionmaker
4. SessionLocal = sessionmaker(
5.     bind=engine,
6.     autoflush=False,
7.     autocommit=False,
8.     expire_on_commit=False,
9. )

复制代码

四、定义 ORM 模型
首先创建基类：

2. from sqlalchemy.orm import DeclarativeBase
4. class Base(DeclarativeBase):
5.     pass

复制代码

接着定义业务模型，以 User 为例：

2. from datetime import datetime
3. from sqlalchemy import DateTime, Integer, String, func
4. from sqlalchemy.orm import Mapped, mapped_column
5. from app.db.base import Base
7. class User(Base):
8.     __tablename__ = "users"
10.     id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
11.     username: Mapped[str] = mapped_column(String(50), unique=True, nullable=False)
12.     email: Mapped[str] = mapped_column(String(191), unique=True, nullable=False)
13.     status: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
14.     created_at: Mapped[datetime] = mapped_column(
15.         DateTime(timezone=True),
16.         server_default=func.now(),
17.         nullable=False,
18.     )

复制代码

五、增删改查实现
1. 查询

2. from sqlalchemy import select
3. from sqlalchemy.orm import Session
5. def list_users(db: Session) -> list[User]:
6.     return list(db.scalars(select(User)).all())
8. def get_user(db: Session, user_id: int) -> User | None:
9.     return db.get(User, user_id)
11. def get_user_by_username(db: Session, username: str) -> User | None:
12.     return db.scalar(select(User).where(User.username == username))
14. def page_users(db: Session, page: int, page_size: int) -> dict:
15.     offset = (page - 1) * page_size
16.     total = db.scalar(select(func.count()).select_from(User)) or 0
17.     items = db.scalars(
18.         select(User).order_by(User.id.desc()).offset(offset).limit(page_size)
19.     ).all()
20.     return {
21.         "items": list(items),
22.         "total": total,
23.         "page": page,
24.         "page_size": page_size,
25.     }

复制代码

2. 新增

2. def create_user(db: Session, username: str, email: str) -> User:
3.     user = User(username=username, email=email, status=1)
4.     db.add(user)
5.     db.commit()
6.     db.refresh(user)
7.     return user

复制代码

注意：commit 后必须 refresh 才能拿到数据库生成的自增 ID 和默认时间戳。

3. 更新

2. def update_user_email(db: Session, user_id: int, email: str) -> User | None:
3.     user = db.get(User, user_id)
4.     if user is None:
5.         return None
6.     user.email = email
7.     db.commit()
8.     db.refresh(user)
9.     return user

复制代码

直接从会话查出的对象，修改字段后 commit 即可生成 UPDATE。

4. 删除

2. def delete_user(db: Session, user_id: int) -> bool:
3.     user = db.get(User, user_id)
4.     if user is None:
5.         return False
6.     db.delete(user)
7.     db.commit()
8.     return True

复制代码

六、事务管理
写操作建议用 try/commit/rollback 包裹，或者使用 sessionmaker 的 begin 上下文：

2. from sqlalchemy.orm import Session
4. def create_user_with_role(db: Session, user: User, role: Role) -> User:
5.     try:
6.         db.add(user)
7.         db.add(role)
8.         db.commit()
9.         db.refresh(user)
10.         return user
11.     except Exception:
12.         db.rollback()
13.         raise

复制代码

等价写法：

2. with SessionLocal.begin() as db:
3.     db.add(user)
4.     db.add(role)

复制代码

七、在 FastAPI 中注入 Session
每个请求应获取独立会话，依赖注入是最佳实践：

2. from collections.abc import Generator
3. from sqlalchemy.orm import Session
4. from app.db.session import SessionLocal
6. def get_db() -> Generator[Session, None, None]:
7.     db = SessionLocal()
8.     try:
9.         yield db
10.     finally:
11.         db.close()

复制代码

路由中使用：

2. from typing import Annotated
3. from fastapi import APIRouter, Depends
4. from sqlalchemy.orm import Session
5. from app.dependencies.database import get_db
7. router = APIRouter(prefix="/users")
9. @router.get("/")
10. def get_users(db: Session = Depends(get_db)):
11.     # ...

复制代码

也可以定义类型别名：

2. SessionDep = Annotated[Session, Depends(get_db)]

复制代码

八、推荐工程结构
一个可维护的目录布局示例：

2. app/
3. ├── core/
4. │   └── config.py          # 配置读取
5. ├── db/
6. │   ├── base.py            # DeclarativeBase
7. │   ├── session.py         # engine 和 SessionLocal
8. │   └── models.py          # 汇总导入所有 ORM 模型
9. ├── dependencies/
10. │   └── database.py        # get_db 依赖
11. ├── modules/
12. │   └── users/
13. │       ├── models.py      # User ORM 模型
14. │       ├── schemas.py     # Pydantic 数据模型
15. │       ├── repository.py  # 数据库查询封装
16. │       ├── service.py     # 业务逻辑
17. │       └── router.py      # 路由定义
18. ├── api/
19. │   └── v1/
20. │       └── router.py      # 聚合所有模块路由
21. └── main.py                # FastAPI 应用入口

复制代码

九、常见错误排查
1. 忘记提交事务：写操作完成后必须调用 db.commit()，否则数据不会写入。
2. 异常后未回滚：捕获异常后应调用 db.rollback() 释放会话状态，否则后续查询可能报“Pending rollback”错误。
3. 在路由中直接创建 Engine：每个路由函数都创建新的 Engine 会导致连接泄漏，应将 Engine 作为全局单例。
4. 密码硬编码：数据库密码应放在环境变量或 .env 文件中，切勿写死在代码仓库。
5. 模型未被导入：如果使用 Base.metadata.create_all 自动建表，必须确保所有模型模块被提前导入，否则对应表不会创建。

全文围绕 FastAPI + SQLAlchemy + PyMySQL 构建 MySQL 后端，覆盖配置、模型、CRUD、事务与依赖注入，适合作为工程初始化的基础参考。

热心网友2

帖子写得很详细，从依赖安装到环境配置、ORM定义再到增删改查，逻辑清晰，操作性强。特别是用 pydantic-settings 读取 .env 的写法很规范，避免硬编码。有一点想确认：同步版中 SessionLocal 的 `expire_on_commit=False` 是为了避免 commit 后对象过期导致的懒加载问题吗？我遇到过类似场景，这样设置后确实方便很多。 另外，如果后续要升级到异步版（比如用 asyncmy 或 aiomysql），结构和这里类似，只是 Engine 和 Session 的创建方式不同。期待楼主后续能补一个异步版本的对比。

热心网友2

这个教程写得非常清晰完整！从依赖安装、环境配置到模型定义、CRUD实现，每一步都解释得很到位。特别是 `pool_pre_ping=True` 和 `expire_on_commit=False` 这两个参数，很多人容易忽略，能有效避免连接残留问题。另外用 pydantic-settings 读 .env 的方式也很规范。 想请问一下，如果项目中需要写事务（比如同时更新多张表），在 FastAPI 依赖注入里通常会怎么处理呢？是用 `db.begin()` 显式控制，还是依赖 Session 的自动提交/回滚机制？期待能补充一下事务协作方面的示例，谢谢！

热心网友2

感谢楼主的详细梳理！从依赖安装到环境配置再到模型定义，每一步都很清晰，尤其是 `pool_pre_ping` 和 `expire_on_commit` 这些细节，对生产环境很有参考价值。不过注意到你只展示了查询部分的代码，想问一下：在同步版中，你通常如何组织增删改操作？比如使用 `db.add`、`db.commit`、`db.refresh` 时，事务提交和异常回滚的逻辑是放在 Service 层还是通过中间件统一管理？另外 FastAPI 的请求上下文中依赖注入 Session 的方式是否也有示例？期待后续的完整 CRUD 和项目结构分享～