FastAPI 应用向 MySQL 插入数据后未生效，通常是因为缺少数据库事务提交（commit()）操作；本文详解问题成因、修复方法，并提供基于 SQLAlchemy 的健壮实践方案。
fastapi 应用向 mysql 插入数据后未生效，通常是因为缺少数据库事务提交（`commit()`）操作；本文详解问题成因、修复方法，并提供基于 sqlalchemy 的健壮实践方案。
在 FastAPI 中直接使用原生 MySQL 连接（如 mysql-connector-python 或 PyMySQL）执行 INSERT 语句时，一个极易被忽略却至关重要的环节是：显式提交事务。数据库驱动默认以事务模式运行——即使 execute() 成功执行了 SQL，若未调用 commit()，所有变更仅暂存于当前事务上下文中，程序结束或连接关闭时将自动回滚，导致“无错误但无数据”的现象。这正是你遇到的问题核心。
? 问题定位：缺失 commit 与资源管理缺陷
你当前的代码片段中：
mycursor.execute(sql_query, values)
# ❌ 缺少 mycursor.commit()
# ❌ 缺少连接/游标释放逻辑（易引发连接泄漏）
return {"message": "Usuario Insertado"}虽然 execute() 成功返回，但因未提交事务，MySQL 实际未落盘；同时未关闭游标（mycursor.close()）或连接，长期运行可能导致连接池耗尽。
✅ 正确做法：每次写操作后必须 commit()，并确保资源安全释放（推荐使用 try...finally 或上下文管理器）。
✅ 快速修复（原生 MySQL 方案）
以下为修正后的最小可行实现（以 mysql-connector-python 为例）：
from mysql.connector import connect
from mysql.connector.errors import Error
# 假设已建立全局连接（生产环境应使用连接池）
db_config = {
"host": "localhost",
"user": "your_user",
"password": "your_pass",
"database": "your_db",
"charset": "utf8mb4"
}
conn = connect(**db_config)
@app.post("/add_user")
def add_user(user: User):
cursor = conn.cursor()
try:
values = (
user.NOMBRE,
user.APELLIDOS,
user.USUARIO,
user.EMAIL,
user.NACIMIENTO,
user.SEXO,
user.ALTURA,
user.PESO
)
sql_query = """
INSERT INTO Usuarios (Nombre, Apellidos, Usuario, Email, Nacimiento, Sexo, Altura, Peso)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
"""
cursor.execute(sql_query, values)
conn.commit() # ✅ 关键：提交事务
return {"message": "Usuario Insertado", "id": cursor.lastrowid}
except Error as e:
conn.rollback() # ❗ 发生异常时回滚，避免脏数据
raise HTTPException(status_code=500, detail=f"Database error: {e}")
finally:
cursor.close() # ✅ 安全释放游标⚠️ 重要注意事项
- 不要在全局作用域创建单个连接供多请求复用：MySQL 连接非线程安全，高并发下易出错。应使用连接池（如 mysql-connector-python 的 pool_name）或改用 SQLAlchemy。
- 日期/时间类型需匹配：确保 user.NACIMIENTO: datetime.date 与 MySQL DATE 字段兼容（本例正确）；若字段为 DATETIME，则需传入 datetime.datetime。
- SQL 注入风险：你已正确使用参数化查询（%s 占位符），这是安全的；切勿拼接字符串构造 SQL。
-
HTTP 状态码：成功插入建议返回 201 Created 状态码，并附带 Location 头指向新资源，符合 REST 规范：
return Response(content='{"message": "Usuario Insertado"}', status_code=201, media_type="application/json")
? 推荐进阶方案：集成 SQLAlchemy + FastAPI
原生驱动适合学习，但生产环境强烈推荐 SQLAlchemy —— 它提供 ORM 抽象、连接池管理、自动事务控制及更佳的错误处理能力。FastAPI 官方教程也以此为基础：
from sqlalchemy import create_engine, Column, Integer, String, Date, Float, Enum
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from fastapi import Depends, HTTPException
Base = declarative_base()
class UsuarioDB(Base):
__tablename__ = "Usuarios"
id = Column(Integer, primary_key=True, index=True)
Nombre = Column(String(100))
Apellidos = Column(String(100))
Usuario = Column(String(50), unique=True)
Email = Column(String(100))
Nacimiento = Column(Date)
Sexo = Column(String(10))
Altura = Column(Float)
Peso = Column(Float)
# 创建引擎（启用连接池）
engine = create_engine(
"mysql+mysqlconnector://user:pass@localhost/dbname",
pool_pre_ping=True, # 自动验证连接有效性
pool_recycle=3600
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.post("/add_user", status_code=201)
def add_user(user: User, db: Session = Depends(get_db)):
db_user = UsuarioDB(
Nombre=user.NOMBRE,
Apellidos=user.APELLIDOS,
Usuario=user.USUARIO,
Email=user.EMAIL,
Nacimiento=user.NACIMIENTO,
Sexo=user.SEXO,
Altura=user.ALTURA,
Peso=user.PESO
)
db.add(db_user)
db.commit() # ✅ SQLAlchemy 自动管理事务
db.refresh(db_user)
return {"message": "Usuario Insertado", "id": db_user.id}? 优势总结：自动连接池、声明式模型、事务封装、依赖注入式会话管理、更好的可测试性与可维护性。
✅ 总结
| 问题根源 | 解决方案 |
|---|---|
| 未调用 commit() | 在 execute() 后立即 conn.commit() |
| 游标未关闭 | finally 块中 cursor.close() |
| 连接管理粗放 | 改用 SQLAlchemy 连接池 + 依赖注入 |
| 缺乏错误回滚 | try/except 中 conn.rollback() |
记住：数据库写操作 = execute + commit + resource cleanup。从今天起，在每个 FastAPI 数据写入端点中检查这三要素，即可规避 90% 的“数据不落地”问题。
