SQLAlchemy 2.x 连接 MySQL 必须用 mysql+pymysql:// 格式并指定 charset=utf8mb4,生产环境需配置 pool_pre_ping=True 和 pool_recycle=3600,FastAPI 中应通过依赖注入提供 Session 而非 Engine。

SQLAlchemy 2.x 连接 MySQL 的 URL 格式必须带 +pymysql
FastAPI 本身不处理数据库连接,靠 SQLAlchemy 管理;而 SQLAlchemy 默认不内置 MySQL 驱动,必须显式指定 DBAPI。直接用 mysql:// 会报错:ModuleNotFoundError: No module named 'MySQLdb'(它默认找 MySQLdb,但该库不支持 Python 3.12+ 且难装)。正确写法是用 mysql+pymysql:// 或 mysql+mysqldb:// —— 后者基本弃用,所以统一选 pymysql。
实操建议:
立即学习“Python免费学习笔记(深入)”;
调用 Cutout.Pro 视觉处理 API 进行背景移除、人像抠图和照片增强,支持文件上传与图片 URL 输入。
- 安装依赖:
pip install sqlalchemy pymysql python-dotenv - 连接字符串示例:
mysql+pymysql://root:password@127.0.0.1:3306/mydb?charset=utf8mb4 -
charset=utf8mb4必须加,否则 emoji 和部分中文可能乱码或插入失败 - 密码含特殊字符(如
@、/)时,要用urllib.parse.quote_plus()编码后再拼进 URL
用 create_engine() 时要关掉 echo=True 上生产
echo=True 会在终端狂打 SQL 日志,开发调试有用,但上线后不仅暴露查询逻辑,还拖慢性能(I/O + 字符串格式化开销)。更关键的是,它默认开启 pool_pre_ping=False,连接池里的空闲连接可能已断开却没被及时剔除,导致后续请求抛出 sqlalchemy.exc.InvalidRequestError: This Session... is in 'prepared' state 或底层 OperationalError: (2013, 'Lost connection to MySQL server during query')。
实操建议:
立即学习“Python免费学习笔记(深入)”;
- 开发环境:
create_engine(url, echo=True, pool_pre_ping=True) - 生产环境:
create_engine(url, echo=False, pool_pre_ping=True, pool_recycle=3600) -
pool_recycle=3600强制每小时重连一次,避开 MySQL 默认的wait_timeout=28800(8 小时)导致的静默断连 - 别用
connect()手动建连接——FastAPI 生命周期里应由依赖注入管理Session
FastAPI 依赖注入中返回 Session 而非 Engine
每次 HTTP 请求都需要一个独立、线程安全的数据库会话,而不是复用全局 Engine。直接把 Engine 注入路由函数会导致事务混乱、连接泄漏,甚至并发下数据错乱。
实操建议:
立即学习“Python免费学习笔记(深入)”;
- 定义依赖函数:
def get_db(): db = SessionLocal() try: yield db finally: db.close() - 在路由中声明:
def read_user(db: Session = Depends(get_db)) -
SessionLocal由sessionmaker(bind=engine)创建,不是全局单例 - 别在依赖里做
db.commit()—— 事务边界应由业务逻辑控制,FastAPI 依赖只管生命周期
Pydantic v2 模型与 SQLAlchemy 模型混用时字段名对不上
SQLAlchemy 模型字段名(如 user_name)和 Pydantic 输出字段(如 userName)不一致,直接用 model_dump() 会漏字段或报错。FastAPI 默认用 Pydantic v2,其 from_orm() 已废弃,改用 model_validate(),但要求字段名严格匹配。
实操建议:
立即学习“Python免费学习笔记(深入)”;
- SQLAlchemy 模型字段命名尽量用小写+下划线(
created_at),Pydantic 模型也保持一致 - 若需驼峰输出,用 Pydantic 的
model_config = ConfigDict(alias_generator=lambda x: re.sub(r'_([a-z])', lambda m: m.group(1).upper(), x)) - 避免用
__table_args__ = {"schema": "xxx"}同时又没在Base.metadata.create_all()里指定 schema,否则建表失败但无提示
pool_pre_ping 和 pool_recycle 组合没配对,服务跑两天就 intermittent failure,查日志还全是超时或断连,得翻 SQLAlchemy 文档里犄角旮旯的 note 才能定位。

















