基于 SQLAlchemy 2.x面向 MySQL 数据库的从入门到精通实战教程1基于 SQLAlchemy 2.x面向 MySQL 数据库的从入门到精通实战教程2文章目录第10章 SQLAlchemy Core10.1 Core 与 ORM 的区别10.2 Table 定义10.3 SQL 表达式语言10.4 执行原生 SQL第11章 数据库迁移 Alembic11.1 Alembic 简介11.2 初始化与配置11.3 生成与执行迁移11.4 常用命令第12章 MySQL 最佳实践与常见问题12.1 性能优化技巧12.2 MySQL 常见陷阱与解决方案12.3 调试与日志12.4 与 FastAPI 集成MySQL 版12.5 MySQL 字符集与排序规则附录 常用操作速查表结语第10章 SQLAlchemy Core10.1 Core 与 ORM 的区别对比维度ORMCore抽象层级高面向对象中SQL 表达式操作方式通过模型类和对象通过 Table 和 SQL 表达式状态跟踪自动跟踪对象状态变化无状态跟踪关系映射支持 relationship需手动写 JOIN性能有一定开销更接近原生 SQL性能更好适用场景CRUD 为主的业务逻辑复杂查询、批量操作、性能敏感10.2 Table 定义fromsqlalchemyimportTable,Column,Integer,String,MetaData,ForeignKey metadataMetaData()usersTable(users,metadata,Column(id,Integer,primary_keyTrue),Column(name,String(50),nullableFalse),Column(email,String(100),uniqueTrue),)# 创建表metadata.create_all(engine)10.3 SQL 表达式语言fromsqlalchemyimportinsert,select,update,delete# INSERTstmtinsert(users).values(name张三,emailzhangsanexample.com)withengine.connect()asconn:resultconn.execute(stmt)conn.commit()print(result.inserted_primary_key)# SELECTstmtselect(users).where(users.c.name张三)withengine.connect()asconn:rowsconn.execute(stmt).fetchall()forrowinrows:print(row.id,row.name,row.email)# UPDATEstmtupdate(users).where(users.c.id1).values(name新名字)withengine.connect()asconn:resultconn.execute(stmt)conn.commit()print(result.rowcount)# DELETEstmtdelete(users).where(users.c.id1)withengine.connect()asconn:resultconn.execute(stmt)conn.commit()10.4 执行原生 SQLfromsqlalchemyimporttext# Core 风格withengine.connect()asconn:resultconn.execute(text(SELECT * FROM users WHERE age :age),{age:25})rowsresult.fetchall()conn.commit()# ORM Session 中执行withSession(engine)assession:resultsession.execute(text(SELECT * FROM users WHERE name :name),{name:张三})rowsresult.fetchall()# 将原生 SQL 结果映射到 ORM 对象withSession(engine)assession:userssession.execute(select(User).from_statement(text(SELECT * FROM users WHERE age :age)),{age:25}).scalars().all()安全警告执行原生 SQL 时必须使用参数绑定:param占位符切勿使用字符串拼接否则存在 SQL 注入风险。MySQL 专用 SQL 提示# MySQL 索引提示FORCE INDEX / USE INDEXfromsqlalchemyimporttext stmttext(SELECT * FROM users FORCE INDEX (idx_name) WHERE name :name)第11章 数据库迁移 Alembic11.1 Alembic 简介Alembic 是 SQLAlchemy 官方提供的数据库迁移工具用于管理 MySQL 表结构的版本化变更。当模型发生变化时Alembic 可以自动生成迁移脚本并执行。核心功能自动检测模型变化并生成迁移脚本版本化管理支持升级upgrade和降级downgrade迁移脚本可编辑支持自定义数据迁移逻辑11.2 初始化与配置# 安装pipinstallalembic# 初始化alembic init alembic修改配置# alembic.ini sqlalchemy.url mysqlpymysql://root:passlocalhost:3306/mydb?charsetutf8mb4# alembic/env.pyfrommyapp.modelsimportBase target_metadataBase.metadata11.3 生成与执行迁移# 1. 修改模型后自动生成迁移脚本alembic revision--autogenerate-mcreate users table# 2. 检查生成的迁移脚本alembic/versions/ 下的 py 文件# 3. 执行迁移alembic upgradehead# 4. 回退一个版本alembic downgrade-1# 5. 查看当前版本alembic current# 6. 查看迁移历史alembichistory11.4 常用命令命令说明alembic init dir初始化 Alembic 环境alembic revision -m msg创建空迁移脚本alembic revision --autogenerate -m msg根据模型变化自动生成迁移脚本alembic upgrade head升级到最新版本alembic upgrade 1升级一个版本alembic downgrade -1降级一个版本alembic downgrade base降级到初始状态alembic current显示当前数据库版本alembic history显示迁移历史alembic stamp head标记当前数据库为最新版本不执行迁移MySQL 迁移注意自动生成的迁移脚本务必人工检查尤其是列类型、默认值、字符集、排序规则MySQL 的ALTER TABLE在大表上可能很慢且锁表生产环境大表变更建议用pt-online-schema-change或gh-ost迁移脚本生成后应纳入 Git 版本控制生产环境执行迁移前先备份数据库第12章 MySQL 最佳实践与常见问题12.1 性能优化技巧N1 查询问题ORM 最常见的性能陷阱fromsqlalchemy.ormimportselectinload,joinedload# 错误N1 查询userssession.execute(select(User)).scalars().all()foruserinusers:print(user.posts)# 每次访问都产生一次查询# 正确selectinload 预加载一对多推荐stmtselect(User).options(selectinload(User.posts))userssession.execute(stmt).scalars().all()# joinedload 适用于多对一/一对一stmtselect(Post).options(joinedload(Post.author))postssession.execute(stmt).scalars().all()其他 MySQL 性能优化使用 Core 批量操作替代 ORM 逐条操作只查询需要的列select(User.name, User.email)而非select(User)合理设置连接池参数避免连接泄漏对常用查询字段、外键、排序字段添加 MySQL 索引使用session.bulk_save_objects()进行大批量插入大结果集使用.yield_per()分批加载避免内存溢出深分页使用游标分页WHERE id last_id替代OFFSET避免在 MySQL 中使用SELECT *只查需要的列12.2 MySQL 常见陷阱与解决方案问题原因解决方案Lost connection to MySQL serverMySQLwait_timeout断开空闲连接设置pool_pre_pingTruepool_recycle1800MySQL server has gone away连接被服务端关闭或网络中断同上检查网络和 MySQL 配置Packet for query is too largeSQL 或数据超过max_allowed_packet调大 MySQL 的max_allowed_packet或分批插入Deadlock found并发事务交叉锁导致死锁统一加锁顺序缩短事务时间重试机制Lock wait timeout exceeded行锁等待超时优化 SQL缩短事务检查长事务N1 查询懒加载导致循环中多次查询使用selectinload/joinedload预加载中文乱码 / emoji 存不进字符集不是 utf8mb4连接字符串加charsetutf8mb4表和列用utf8mb4DetachedInstanceError访问已脱离 Session 的对象属性Session 关闭前预加载数据或重新查询并发问题多线程共享同一个 Session每个线程/请求使用独立 Session自增主键不连续事务回滚后自增 ID 不回退MySQL 正常行为不影响使用不要依赖 ID 连续性expire_on_commit导致额外查询commit 后对象属性被标记过期设置expire_on_commitFalse或用session.refresh()12.3 调试与日志# 方式一创建 Engine 时设置 echoenginecreate_engine(url,echoTrue)# 方式二通过 logging 精细控制importlogging logging.basicConfig()logging.getLogger(sqlalchemy.engine).setLevel(logging.INFO)# 打印 SQLlogging.getLogger(sqlalchemy.engine).setLevel(logging.DEBUG)# 打印 SQL 参数 结果# 只打印 SQL 不打印结果集logging.getLogger(sqlalchemy.engine.Engine).setLevel(logging.INFO)MySQL EXPLAIN 分析慢查询fromsqlalchemyimporttext resultsession.execute(text(EXPLAIN SELECT * FROM users WHERE age 25))forrowinresult:print(row)MySQL 慢查询日志-- 开启慢查询日志SETGLOBALslow_query_logON;SETGLOBALlong_query_time1;# 超过1秒的查询记录SHOWVARIABLESLIKEslow_query_log_file;12.4 与 FastAPI 集成MySQL 版fromfastapiimportFastAPI,Depends,HTTPExceptionfromsqlalchemyimportcreate_engine,selectfromsqlalchemy.ormimportsessionmaker,Session,DeclarativeBase appFastAPI()# MySQL 数据库配置生产环境从环境变量读取SQLALCHEMY_DATABASE_URLmysqlpymysql://root:passlocalhost:3306/mydb?charsetutf8mb4enginecreate_engine(SQLALCHEMY_DATABASE_URL,pool_size10,max_overflow20,pool_pre_pingTrue,pool_recycle1800,)SessionLocalsessionmaker(autocommitFalse,autoflushFalse,bindengine)classBase(DeclarativeBase):pass# 依赖获取数据库 Sessiondefget_db():dbSessionLocal()try:yielddbfinally:db.close()# 路由示例app.post(/users/)defcreate_user(name:str,email:str,db:SessionDepends(get_db)):userUser(namename,emailemail)db.add(user)db.commit()db.refresh(user)returnuserapp.get(/users/{user_id})defget_user(user_id:int,db:SessionDepends(get_db)):userdb.get(User,user_id)ifnotuser:raiseHTTPException(status_code404,detailUser not found)returnuser12.5 MySQL 字符集与排序规则创建数据库时指定字符集CREATEDATABASEmydbCHARACTERSETutf8mb4COLLATEutf8mb4_unicode_ci;SQLAlchemy 中指定表的字符集和引擎classUser(Base):__tablename__users__table_args__{mysql_engine:InnoDB,mysql_charset:utf8mb4,mysql_collate:utf8mb4_unicode_ci,}id:Mapped[int]mapped_column(Integer,primary_keyTrue)name:Mapped[str]mapped_column(String(50),nullableFalse)排序规则选择utf8mb4_general_ci排序较快但准确性一般不支持某些语言的特殊排序utf8mb4_unicode_ci排序更准确基于 Unicode 标准推荐使用utf8mb4_bin二进制排序区分大小写区分重音字符附录 常用操作速查表操作代码创建 Enginecreate_engine(mysqlpymysql://...)创建 SessionSession(engine)定义模型class X(Base): __tablename__ x创建表Base.metadata.create_all(engine)新增session.add(obj); session.commit()批量新增session.add_all(list); session.commit()按主键查session.get(Model, id)查所有session.execute(select(Model)).scalars().all()条件查select(Model).where(Model.field val)排序select(Model).order_by(Model.field.desc())分页select(Model).limit(n).offset(m)更新obj.field val; session.commit()删除session.delete(obj); session.commit()计数func.count(Model.id)分组select(...).group_by(Model.field)关联relationship()外键ForeignKey(table.id)预加载select(Model).options(selectinload(Model.rel))事务with session.begin(): ...回滚session.rollback()原生 SQLsession.execute(text(SQL), params)生成迁移alembic revision --autogenerate -m msg执行迁移alembic upgrade head结语SQLAlchemy MySQL 是 Python 后端开发中最经典的组合之一。掌握它的核心概念——Engine、Session、Model、relationship、transaction——是高效使用的基础。在 MySQL 场景下特别需要注意连接字符串务必指定charsetutf8mb4生产环境必设pool_pre_pingTruepool_recycle1800防止断连金额用Numeric主键用BigInteger长文本用LONGTEXT注意 N1 查询合理使用selectinload预加载使用 Alembic 管理数据库迁移不要手动改表结构大表 DDL 变更注意锁表问题考虑用 online schema change 工具Web 应用中使用请求级别的 Session避免共享更多详细信息请参考官方文档SQLAlchemy 官方文档https://docs.sqlalchemy.org/Alembic 官方文档https://alembic.sqlalchemy.org/MySQL 官方文档https://dev.mysql.com/doc/注文档部分内容可能由 AI 生成 基于 SQLAlchemy 2.x，面向 MySQL 数据库的从入门到精通实战教程（3）
基于 SQLAlchemy 2.x面向 MySQL 数据库的从入门到精通实战教程1基于 SQLAlchemy 2.x面向 MySQL 数据库的从入门到精通实战教程2文章目录第10章 SQLAlchemy Core10.1 Core 与 ORM 的区别10.2 Table 定义10.3 SQL 表达式语言10.4 执行原生 SQL第11章 数据库迁移 Alembic11.1 Alembic 简介11.2 初始化与配置11.3 生成与执行迁移11.4 常用命令第12章 MySQL 最佳实践与常见问题12.1 性能优化技巧12.2 MySQL 常见陷阱与解决方案12.3 调试与日志12.4 与 FastAPI 集成MySQL 版12.5 MySQL 字符集与排序规则附录 常用操作速查表结语第10章 SQLAlchemy Core10.1 Core 与 ORM 的区别对比维度ORMCore抽象层级高面向对象中SQL 表达式操作方式通过模型类和对象通过 Table 和 SQL 表达式状态跟踪自动跟踪对象状态变化无状态跟踪关系映射支持 relationship需手动写 JOIN性能有一定开销更接近原生 SQL性能更好适用场景CRUD 为主的业务逻辑复杂查询、批量操作、性能敏感10.2 Table 定义fromsqlalchemyimportTable,Column,Integer,String,MetaData,ForeignKey metadataMetaData()usersTable(users,metadata,Column(id,Integer,primary_keyTrue),Column(name,String(50),nullableFalse),Column(email,String(100),uniqueTrue),)# 创建表metadata.create_all(engine)10.3 SQL 表达式语言fromsqlalchemyimportinsert,select,update,delete# INSERTstmtinsert(users).values(name张三,emailzhangsanexample.com)withengine.connect()asconn:resultconn.execute(stmt)conn.commit()print(result.inserted_primary_key)# SELECTstmtselect(users).where(users.c.name张三)withengine.connect()asconn:rowsconn.execute(stmt).fetchall()forrowinrows:print(row.id,row.name,row.email)# UPDATEstmtupdate(users).where(users.c.id1).values(name新名字)withengine.connect()asconn:resultconn.execute(stmt)conn.commit()print(result.rowcount)# DELETEstmtdelete(users).where(users.c.id1)withengine.connect()asconn:resultconn.execute(stmt)conn.commit()10.4 执行原生 SQLfromsqlalchemyimporttext# Core 风格withengine.connect()asconn:resultconn.execute(text(SELECT * FROM users WHERE age :age),{age:25})rowsresult.fetchall()conn.commit()# ORM Session 中执行withSession(engine)assession:resultsession.execute(text(SELECT * FROM users WHERE name :name),{name:张三})rowsresult.fetchall()# 将原生 SQL 结果映射到 ORM 对象withSession(engine)assession:userssession.execute(select(User).from_statement(text(SELECT * FROM users WHERE age :age)),{age:25}).scalars().all()安全警告执行原生 SQL 时必须使用参数绑定:param占位符切勿使用字符串拼接否则存在 SQL 注入风险。MySQL 专用 SQL 提示# MySQL 索引提示FORCE INDEX / USE INDEXfromsqlalchemyimporttext stmttext(SELECT * FROM users FORCE INDEX (idx_name) WHERE name :name)第11章 数据库迁移 Alembic11.1 Alembic 简介Alembic 是 SQLAlchemy 官方提供的数据库迁移工具用于管理 MySQL 表结构的版本化变更。当模型发生变化时Alembic 可以自动生成迁移脚本并执行。核心功能自动检测模型变化并生成迁移脚本版本化管理支持升级upgrade和降级downgrade迁移脚本可编辑支持自定义数据迁移逻辑11.2 初始化与配置# 安装pipinstallalembic# 初始化alembic init alembic修改配置# alembic.ini sqlalchemy.url mysqlpymysql://root:passlocalhost:3306/mydb?charsetutf8mb4# alembic/env.pyfrommyapp.modelsimportBase target_metadataBase.metadata11.3 生成与执行迁移# 1. 修改模型后自动生成迁移脚本alembic revision--autogenerate-mcreate users table# 2. 检查生成的迁移脚本alembic/versions/ 下的 py 文件# 3. 执行迁移alembic upgradehead# 4. 回退一个版本alembic downgrade-1# 5. 查看当前版本alembic current# 6. 查看迁移历史alembichistory11.4 常用命令命令说明alembic init dir初始化 Alembic 环境alembic revision -m msg创建空迁移脚本alembic revision --autogenerate -m msg根据模型变化自动生成迁移脚本alembic upgrade head升级到最新版本alembic upgrade 1升级一个版本alembic downgrade -1降级一个版本alembic downgrade base降级到初始状态alembic current显示当前数据库版本alembic history显示迁移历史alembic stamp head标记当前数据库为最新版本不执行迁移MySQL 迁移注意自动生成的迁移脚本务必人工检查尤其是列类型、默认值、字符集、排序规则MySQL 的ALTER TABLE在大表上可能很慢且锁表生产环境大表变更建议用pt-online-schema-change或gh-ost迁移脚本生成后应纳入 Git 版本控制生产环境执行迁移前先备份数据库第12章 MySQL 最佳实践与常见问题12.1 性能优化技巧N1 查询问题ORM 最常见的性能陷阱fromsqlalchemy.ormimportselectinload,joinedload# 错误N1 查询userssession.execute(select(User)).scalars().all()foruserinusers:print(user.posts)# 每次访问都产生一次查询# 正确selectinload 预加载一对多推荐stmtselect(User).options(selectinload(User.posts))userssession.execute(stmt).scalars().all()# joinedload 适用于多对一/一对一stmtselect(Post).options(joinedload(Post.author))postssession.execute(stmt).scalars().all()其他 MySQL 性能优化使用 Core 批量操作替代 ORM 逐条操作只查询需要的列select(User.name, User.email)而非select(User)合理设置连接池参数避免连接泄漏对常用查询字段、外键、排序字段添加 MySQL 索引使用session.bulk_save_objects()进行大批量插入大结果集使用.yield_per()分批加载避免内存溢出深分页使用游标分页WHERE id last_id替代OFFSET避免在 MySQL 中使用SELECT *只查需要的列12.2 MySQL 常见陷阱与解决方案问题原因解决方案Lost connection to MySQL serverMySQLwait_timeout断开空闲连接设置pool_pre_pingTruepool_recycle1800MySQL server has gone away连接被服务端关闭或网络中断同上检查网络和 MySQL 配置Packet for query is too largeSQL 或数据超过max_allowed_packet调大 MySQL 的max_allowed_packet或分批插入Deadlock found并发事务交叉锁导致死锁统一加锁顺序缩短事务时间重试机制Lock wait timeout exceeded行锁等待超时优化 SQL缩短事务检查长事务N1 查询懒加载导致循环中多次查询使用selectinload/joinedload预加载中文乱码 / emoji 存不进字符集不是 utf8mb4连接字符串加charsetutf8mb4表和列用utf8mb4DetachedInstanceError访问已脱离 Session 的对象属性Session 关闭前预加载数据或重新查询并发问题多线程共享同一个 Session每个线程/请求使用独立 Session自增主键不连续事务回滚后自增 ID 不回退MySQL 正常行为不影响使用不要依赖 ID 连续性expire_on_commit导致额外查询commit 后对象属性被标记过期设置expire_on_commitFalse或用session.refresh()12.3 调试与日志# 方式一创建 Engine 时设置 echoenginecreate_engine(url,echoTrue)# 方式二通过 logging 精细控制importlogging logging.basicConfig()logging.getLogger(sqlalchemy.engine).setLevel(logging.INFO)# 打印 SQLlogging.getLogger(sqlalchemy.engine).setLevel(logging.DEBUG)# 打印 SQL 参数 结果# 只打印 SQL 不打印结果集logging.getLogger(sqlalchemy.engine.Engine).setLevel(logging.INFO)MySQL EXPLAIN 分析慢查询fromsqlalchemyimporttext resultsession.execute(text(EXPLAIN SELECT * FROM users WHERE age 25))forrowinresult:print(row)MySQL 慢查询日志-- 开启慢查询日志SETGLOBALslow_query_logON;SETGLOBALlong_query_time1;# 超过1秒的查询记录SHOWVARIABLESLIKEslow_query_log_file;12.4 与 FastAPI 集成MySQL 版fromfastapiimportFastAPI,Depends,HTTPExceptionfromsqlalchemyimportcreate_engine,selectfromsqlalchemy.ormimportsessionmaker,Session,DeclarativeBase appFastAPI()# MySQL 数据库配置生产环境从环境变量读取SQLALCHEMY_DATABASE_URLmysqlpymysql://root:passlocalhost:3306/mydb?charsetutf8mb4enginecreate_engine(SQLALCHEMY_DATABASE_URL,pool_size10,max_overflow20,pool_pre_pingTrue,pool_recycle1800,)SessionLocalsessionmaker(autocommitFalse,autoflushFalse,bindengine)classBase(DeclarativeBase):pass# 依赖获取数据库 Sessiondefget_db():dbSessionLocal()try:yielddbfinally:db.close()# 路由示例app.post(/users/)defcreate_user(name:str,email:str,db:SessionDepends(get_db)):userUser(namename,emailemail)db.add(user)db.commit()db.refresh(user)returnuserapp.get(/users/{user_id})defget_user(user_id:int,db:SessionDepends(get_db)):userdb.get(User,user_id)ifnotuser:raiseHTTPException(status_code404,detailUser not found)returnuser12.5 MySQL 字符集与排序规则创建数据库时指定字符集CREATEDATABASEmydbCHARACTERSETutf8mb4COLLATEutf8mb4_unicode_ci;SQLAlchemy 中指定表的字符集和引擎classUser(Base):__tablename__users__table_args__{mysql_engine:InnoDB,mysql_charset:utf8mb4,mysql_collate:utf8mb4_unicode_ci,}id:Mapped[int]mapped_column(Integer,primary_keyTrue)name:Mapped[str]mapped_column(String(50),nullableFalse)排序规则选择utf8mb4_general_ci排序较快但准确性一般不支持某些语言的特殊排序utf8mb4_unicode_ci排序更准确基于 Unicode 标准推荐使用utf8mb4_bin二进制排序区分大小写区分重音字符附录 常用操作速查表操作代码创建 Enginecreate_engine(mysqlpymysql://...)创建 SessionSession(engine)定义模型class X(Base): __tablename__ x创建表Base.metadata.create_all(engine)新增session.add(obj); session.commit()批量新增session.add_all(list); session.commit()按主键查session.get(Model, id)查所有session.execute(select(Model)).scalars().all()条件查select(Model).where(Model.field val)排序select(Model).order_by(Model.field.desc())分页select(Model).limit(n).offset(m)更新obj.field val; session.commit()删除session.delete(obj); session.commit()计数func.count(Model.id)分组select(...).group_by(Model.field)关联relationship()外键ForeignKey(table.id)预加载select(Model).options(selectinload(Model.rel))事务with session.begin(): ...回滚session.rollback()原生 SQLsession.execute(text(SQL), params)生成迁移alembic revision --autogenerate -m msg执行迁移alembic upgrade head结语SQLAlchemy MySQL 是 Python 后端开发中最经典的组合之一。掌握它的核心概念——Engine、Session、Model、relationship、transaction——是高效使用的基础。在 MySQL 场景下特别需要注意连接字符串务必指定charsetutf8mb4生产环境必设pool_pre_pingTruepool_recycle1800防止断连金额用Numeric主键用BigInteger长文本用LONGTEXT注意 N1 查询合理使用selectinload预加载使用 Alembic 管理数据库迁移不要手动改表结构大表 DDL 变更注意锁表问题考虑用 online schema change 工具Web 应用中使用请求级别的 Session避免共享更多详细信息请参考官方文档SQLAlchemy 官方文档https://docs.sqlalchemy.org/Alembic 官方文档https://alembic.sqlalchemy.org/MySQL 官方文档https://dev.mysql.com/doc/注文档部分内容可能由 AI 生成