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 会挂起但不执行，请求永远不返回。
Mysql Skill 1.0.0 (3)
下载
MySQL 数据库管理技能，能够通过自然语言查询和管理数据库，支持 SELECT/INSERT/UPDATE/DELETE、表管理、备份恢复等操作。当用户提及 MySQL、数据库查询、建表或数据备份时使用此技能。
- 异步驱动二选一：
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 构造器自动生成参数化语句。
- 查单条：
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 报错不会告诉你缺这步。
