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% 的“数据不落地”问题。