FastAPI连接MySQL应优先选同步方案(pymysql+SQLAlchemy),因其稳定、易调试;异步需用aiomysql或asyncmy配合AsyncEngine和async_sessionmaker,ORM操作须全部await,且建表需显式调用create_all。

FastAPI 连接 MySQL 并实现增删改查,关键不在于“能不能”,而在于选同步还是异步——选错驱动或会话模式,get_db 一注入就报 RuntimeError: There is no current event loop in thread,或者接口卡死、并发上不去。
用 pymysql + 同步 SQLAlchemy 最稳,适合中小项目
多数新手卡在第一步:装了 aiomysql 却用同步写法,或反过来。同步方案不需要改事件循环,兼容性好,调试直观。
-
pymysql是纯 Python 实现的 MySQL 驱动,sqlalchemy默认就认它;装命令只要pip install sqlalchemy pymysql - 连接 URL 格式必须是
mysql+pymysql://user:password@host:port/dbname,漏掉+pymysql会报ModuleNotFoundError: No module named 'MySQLdb' -
create_engine要加echo=True(开发期)和pool_pre_ping=True(生产期),否则连接池空闲超时后首次查询直接炸 - 依赖注入函数
get_db必须用yield,且db.close()放在finally块里,否则连接泄漏
async/await 写法必须配 aiomysql 或 asyncmy
想用 async def create_item(...),就不能用 pymysql ——它不支持 await。强行 await 会挂起但不执行,请求永远不返回。
- 异步驱动二选一:
aiomysql(基于 pymysql 异步封装)或asyncmy(纯异步重写,更轻更快)。装法:pip install "sqlalchemy[asyncio]" aiomysql - 引擎 URL 改为
mysql+aiomysql://...,且create_async_engine返回的是AsyncEngine,不能传给同步的sessionmaker - 会话工厂必须用
async_sessionmaker,依赖注入得写成async def get_db() -> AsyncSession,里面用async with - 所有 ORM 操作都要
await:await session.execute(...)、await session.commit(),漏一个就同步阻塞
增删改查别硬写 SQL,用 ORM 的 query API 更安全
直接拼接字符串查 "SELECT * FROM users WHERE id = " + str(user_id) 是 SQL 注入温床。SQLAlchemy 的 select / delete 构造器自动生成参数化语句。
MySQL 数据库管理技能,能够通过自然语言查询和管理数据库,支持 SELECT/INSERT/UPDATE/DELETE、表管理、备份恢复等操作。当用户提及 MySQL、数据库查询、建表或数据备份时使用此技能。
- 查单条:
stmt = select(User).where(User.id == user_id)→result = await session.execute(stmt)→user = result.scalar_one_or_none() - 新增:
user = User(name="Alice")→session.add(user)→await session.flush()(获取自增 ID)→await session.commit() - 删改前务必
await session.get(User, id)先查出来,再改属性或调session.delete(),不要用UPDATE ... WHERE原生语句——事务控制会丢 - 批量插入慎用
session.add_all(),大数据量时改用session.execute(insert(User), [...]),避免内存爆掉
环境变量和错误处理不能省
本地跑通不等于上线能用。DB_PASSWORD 硬编码在代码里,CI/CD 一跑就崩;没捕获 NoResultFound,404 直接变 500。
- 必须用
python-dotenv加载.env文件,SQLALCHEMY_DATABASE_URL从os.getenv()读,且.env加进.gitignore -
数据库连接失败要兜底:在
get_db里 catchOperationalError,返回HTTPException(status_code=503) - 查不到数据统一用
scalar_one_or_none()+ 判空,别用first()——它不抛异常,容易埋逻辑 bug - 事务结束必须显式
commit()或rollback(),异步下session.close()不等于事务结束
最常被忽略的一点:模型类定义后,必须调用 Base.metadata.create_all(engine)(同步)或 await async_engine.run_sync(Base.metadata.create_all)(异步)建表,否则 Table not found 报错不会告诉你缺这步。
大量免费API接口:立即使用
涵盖生活服务API、金融科技API、企业工商API、等相关的API接口服务。免费API接口可安全、合规地连接上下游,为数据API应用能力赋能!










