FastAPI + SQLAlchemy + PyMySQL 连接 My

脚本专家 发表于 4 天前

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

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

pip install "fastapi" sqlalchemy pymysql pydantic-settings

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

pip install "pymysql"

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

pip install alembic

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

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

用 pydantic-settings 读取：

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
model_config = SettingsConfigDict(
   env_file=".env",
   env_file_encoding="utf-8",
)
database_url: str
database_echo: bool = False
database_auto_create: bool = False

settings = Settings()

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

from sqlalchemy import create_engine
from app.core.config import settings

engine = create_engine(
settings.database_url,
echo=settings.database_echo,
pool_pre_ping=True, # 取出连接前检查是否存活
)

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

from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(
bind=engine,
autoflush=False,
autocommit=False,
expire_on_commit=False,
)

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

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
pass

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

from datetime import datetime
from sqlalchemy import DateTime, Integer, String, func
from sqlalchemy.orm import Mapped, mapped_column
from app.db.base import Base

class User(Base):
__tablename__ = "users"

id: Mapped = mapped_column(Integer, primary_key=True, autoincrement=True)
username: Mapped = mapped_column(String(50), unique=True, nullable=False)
email: Mapped = mapped_column(String(191), unique=True, nullable=False)
status: Mapped = mapped_column(Integer, nullable=False, default=1)
created_at: Mapped = mapped_column(
   DateTime(timezone=True),
   server_default=func.now(),
   nullable=False,
)

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

from sqlalchemy import select
from sqlalchemy.orm import Session

def list_users(db: Session) -> list:
return list(db.scalars(select(User)).all())

def get_user(db: Session, user_id: int) -> User | None:
return db.get(User, user_id)

def get_user_by_username(db: Session, username: str) -> User | None:
return db.scalar(select(User).where(User.username == username))

def page_users(db: Session, page: int, page_size: int) -> dict:
offset = (page - 1) * page_size
total = db.scalar(select(func.count()).select_from(User)) or 0
items = db.scalars(
   select(User).order_by(User.id.desc()).offset(offset).limit(page_size)
).all()
return {
   "items": list(items),
   "total": total,
   "page": page,
   "page_size": page_size,
}

2. 新增

def create_user(db: Session, username: str, email: str) -> User:
user = User(username=username, email=email, status=1)
db.add(user)
db.commit()
db.refresh(user)
return user

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

3. 更新

def update_user_email(db: Session, user_id: int, email: str) -> User | None:
user = db.get(User, user_id)
if user is None:
   return None
user.email = email
db.commit()
db.refresh(user)
return user

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

4. 删除

def delete_user(db: Session, user_id: int) -> bool:
user = db.get(User, user_id)
if user is None:
   return False
db.delete(user)
db.commit()
return True

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

from sqlalchemy.orm import Session

def create_user_with_role(db: Session, user: User, role: Role) -> User:
try:
   db.add(user)
   db.add(role)
   db.commit()
   db.refresh(user)
   return user
except Exception:
   db.rollback()
   raise

等价写法：

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

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

from collections.abc import Generator
from sqlalchemy.orm import Session
from app.db.session import SessionLocal

def get_db() -> Generator:
db = SessionLocal()
try:
   yield db
finally:
   db.close()

路由中使用：

from typing import Annotated
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from app.dependencies.database import get_db

router = APIRouter(prefix="/users")

@router.get("/")
def get_users(db: Session = Depends(get_db)):
# ...

也可以定义类型别名：

SessionDep = Annotated

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

app/
├── core/
│ └── config.py       # 配置读取
├── db/
│ ├── base.py          # DeclarativeBase
│ ├── session.py       # engine 和 SessionLocal
│ └── models.py       # 汇总导入所有 ORM 模型
├── dependencies/
│ └── database.py    # get_db 依赖
├── modules/
│ └── users/
│    ├── models.py    # User ORM 模型
│    ├── schemas.py # Pydantic 数据模型
│    ├── repository.py# 数据库查询封装
│    ├── service.py # 业务逻辑
│    └── router.py    # 路由定义
├── api/
│ └── v1/
│    └── router.py    # 聚合所有模块路由
└── 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 发表于 4 天前

Re: FastAPI + SQLAlchemy + PyMySQL 连接 My

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

热心网友2 发表于 4 天前

Re: FastAPI + SQLAlchemy + PyMySQL 连接 My

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

热心网友2 发表于 4 天前

Re: FastAPI + SQLAlchemy + PyMySQL 连接 My

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

页: [1]

红客联盟 - 由08小组运营's Archiver