前两篇把SQLAlchemy的环境配置、模型映射和连接池管理都过了一遍这篇直接进入正题——用SQLAlchemy完成增删改查CRUD。这是所有业务系统里最基础、最高频的操作也是很多人入门ORM后第一道必须跨过的门槛。我尽量把每个操作背后的逻辑讲透比如为什么推荐先查再改、flush和commit到底差在哪、批量操作怎么写才优雅最后附上我实际跑项目时踩过的一些坑。1. 增删改查的整体思路与操作前置准备1.1 为什么单独把CRUD拎出来讲很多教程把增删改查分散在环境搭建和模型定义里看起来省篇幅实际上对新手很不友好。CRUD并不是简单的“换成SQL语法写一遍”里面有很多ORM特有的机制比如会话状态、主键回填、懒加载触发时机这些不彻底搞清楚写出来的代码往往“能用但味道不对”后期并发一上来就各种报错。我在参与某个Web系统的数据层重构时看到过一段让我印象很深的代码——每次查询都手动创建一个新连接用完也不关数据库连接池直接被拖垮。问题不在技术难度而是对Session的生命周期没有概念。所以这篇会把Session机制作为开篇重点它是理解整个数据库操作入口的钥匙然后逐项拆解增、删、改、查。1.2 一切操作都从Session开始SQLAlchemy 2.x 里ORM操作的核心对象是Session。可以把它理解成一次“工作单元”的载体负责跟踪模型对象的变更状态并在合适的时机把变更同步回数据库。一个Session内部持有数据库连接但它并不等同于连接更准确地说它是在连接之上封装了一层对象状态管理。from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, Session engine create_engine( mysqlpymysql://username:passwordlocalhost:3306/demo?charsetutf8mb4, echoFalse ) SessionLocal sessionmaker(bindengine, autoflushFalse, expire_on_commitFalse)这里有两个参数值得展开。autoflushFalse意味着查询前不会自动把未提交的修改先刷到数据库这在后面讲“先查后改”时至关重要expire_on_commitFalse表示提交后对象上的属性不会被强制过期下次访问属性时不用再发一次SQL对性能敏感的业务很友好。实际项目中我习惯把SessionLocal做成模块级单例再用依赖注入或上下文管理器传给业务函数。1.3 ORM操作和原生SQL的映射关系初学者最容易懵的一点明明写的是Python对象操作怎么就到数据库里变成了一条SQL其实每一步操作都有对应的SQL语义。ORM操作底层SQL语义关键方法/属性新增对象INSERT INTO ...session.add(obj)/session.commit()按主键查询SELECT ... WHERE id ?session.get(Model, pk)条件筛选SELECT ... WHERE ...session.query(Model).filter(...)修改对象属性UPDATE ... SET ...先查后改 /session.query(Model).update(...)删除对象DELETE FROM ...session.delete(obj)/query.delete()搞清楚这个映射关系后写CRUD会有一种“心里有底”的感觉。接下来的每一节我都按这个表格展开配合可运行代码和补充参数说明来讲。2. 新增数据从单条到批量的完整姿势2.1 单条新增最简写法假设已经定义好一个User模型包含username、email、created_at三个字段。新增一条用户记录最常规的写法如下with SessionLocal() as session: new_user User( usernamealice, emailaliceexample.com, created_atdatetime.now() ) session.add(new_user) session.commit()有人可能会问add之后不commit行不行不执行commit事务就不会提交数据不会真正写入数据库Session关闭时未提交的修改会被回滚丢掉。commit的行为类似于保存文件——add只是放在了“内存草稿箱”commit才是“点击保存”。2.2 新增后立刻拿到自增主键很多业务场景需要新增后马上用主键做一些后续操作。SQLAlchemy在commit之后会自动把数据库返回的自增ID回填到对象的主键属性上with SessionLocal() as session: new_user User(usernamebob, emailbobexample.com) session.add(new_user) session.commit() user_id new_user.id # 这里已经是数据库里真实的主键值在这个回填机制中其实起作用的是flush。如果等不到commit也可以手动flush来提前触发主键回填——但flush只是把SQL发给数据库事务仍未提交后续出错还可以整体回滚。这也是我比较推荐的一种方式尤其适合需要先拿到ID再关联写入子表数据的场景。session.add(new_user) session.flush() user_id new_user.id # flush后主键已回填2.3 批量新增的两个常用方案一次插入几十上百条数据时逐条add再commit显然太低效。常见方案有两种。第一种是用add_all把对象列表一次性加入会话users [ User(usernamefuser_{i}, emailfuser_{i}example.com) for i in range(100) ] session.add_all(users) session.commit()第二种是直接用session.bulk_insert_mappings绕开完整的ORM状态跟踪速度更快但不会自动回填主键也不会触发模型上的事件监听session.bulk_insert_mappings( User, [ {username: fuser_{i}, email: fuser_{i}example.com} for i in range(100) ] ) session.commit()至于哪种更好我的意见是几十条的量级两者差距不明显优先add_all语义更统一超过几百条再做批量插入时可以考虑bulk_insert_mappings能明显加快整体插入速度。如果再往上走比如一次性导入上万条就该考虑executemany或COPY类方案了。2.4 实操心得主键回填失败的排查方向我在某次批量导入过程中遇到过主键回填失效排查半天发现是模型里主键字段没有设置为autoincrement。MySQL默认对整数主键开启自增但如果你显式定义了primary_keyTrue却没有挂autoincrement有些数据库方言下回填会出问题。经验法则只要主键是单一整数直接写mapped_column(Integer, primary_keyTrue, autoincrementTrue)省心很多。另外如果使用bulk_insert_mappings也别指望它能回填主键需要回填ID时请绕回add/add_all。3. 查询操作filter、条件组合与结果处理查询是CRUD里占比最高的部分也是优化收益最明显的部分。我按“基础查询→条件组合→排序分页→聚合关联”四个层次来讲。这一节内容比较长建议配合代码实际操作一遍。3.1 query.filter与filter_by的区别query(Model)是Query对象构建的起点然后可以在它后面链条式追加过滤条件。两种过滤方法# filter: 接收SQLAlchemy表达式 session.query(User).filter(User.id 10).all() # filter_by: 接收关键字参数只支持等值匹配 session.query(User).filter_by(usernamealice).first()filter的表达能力更强各种、、like、in、or_都靠它filter_by本质是filter(Model.attr value)的简写适合简单等值场景。不要混用太狠保持同一种风格代码读起来更整洁。3.2 常用查询条件速查表需求写法等值匹配User.id 10不等于User.id ! 10模糊匹配User.username.like(%alice%)范围匹配User.age.between(18, 30)包含于集合User.id.in_([1, 2, 3])为空判断User.email.is_(None)/isnot(None)多个条件同时满足(AND)and_(User.id 1, User.age 30)多个条件满足其一(OR)or_(User.id 1, User.username admin)实际使用时我建议优先用and_、or_显式组合可读性更好。也可以直接用Python的和|运算符但要给每个条件加括号否则容易踩运算符优先级的坑例如from sqlalchemy import and_, or_ condition and_( User.age 18, or_(User.username.like(alice%), User.email.like(%example.com)) ) results session.query(User).filter(condition).all()3.3 排序、分页、限量查询排序用order_by支持多个字段组合。分页有两种惯用方式经典limit/offset和按主键游标分页。对于后台管理系统中小数据量的分页limit/offset完全够用一旦数据量过十万offset偏移越大越慢因为数据库要扫描和丢弃前面的行。# 排序 限量 session.query(User).order_by(User.created_at.desc(), User.id.asc()).limit(10).all() # 分页第2页每页20条 page, per_page 2, 20 session.query(User).order_by(User.id).offset((page - 1) * per_page).limit(per_page).all() # 游标分页以last_id为游标取之后20条 session.query(User).filter(User.id last_id).order_by(User.id).limit(per_page).all()limit(0)在SQLAlchemy里会被忽略做“取全部”时要注意这个细节。另外order_by尽量跟随索引走否则分页查询数据量一大很容易拖垮数据库性能。3.4 只查需要的字段避免整行加载不需要全部列时用with_entities或直接用旧式query(Model.字段名)限制返回列这样既省内存又加快查询。我见过不少同事习惯性query(Order).all()然后只取其中两个字段数据量大时白白浪费很多IO。# 方案一查询指定列 rows session.query(User.id, User.username).filter(User.age 18).all() # 方案二查询完整实体但只加载部分字段 rows session.query(User).options(load_only(User.id, User.username)).all()如果你只是想判断“是否存在”正确做法是用session.query(User.id).filter(...).first()而不是查全表然后判断list是否为空。判断存在时也可以直接用exists()from sqlalchemy import exists exists_flag session.query( exists().where(User.email aliceexample.com) ).scalar()3.5 聚合与分组聚合函数是常用能力。统计用户数、按状态分组统计数量、求最大最小值都集中在func里from sqlalchemy import func # 总数 total session.query(func.count(User.id)).scalar() # 按状态分组统计 rows session.query(User.status, func.count(User.id)).group_by(User.status).all() # 数量大于5的状态分组 rows ( session.query(User.status, func.count(User.id).label(cnt)) .group_by(User.status) .having(func.count(User.id) 5) .all() )注意scalar()只取第一行第一列的值聚合查询返回单值时非常方便。分组统计如果要过滤分组一定要用having放到where里会报错或者得到错误语义。3.6 关联查询与加载策略实际项目很少只有单表查询。User和Address、Order和OrderItem这类一对多、多对一关系非常普遍。SQLAlchemy关联查询的用法# 通过relationship属性访问关联对象默认懒加载 user session.query(User).filter(User.id 1).first() addresses user.addresses # 此时才触发额外的SELECT # 显式join查询 rows ( session.query(User, Address) .join(Address, User.id Address.user_id) .filter(Address.city Shanghai) .all() ) # 预先加载关联对象避免N1查询 from sqlalchemy.orm import joinedload users ( session.query(User) .options(joinedload(User.addresses)) .all() )懒加载是很多人容易踩坑的重灾区。在Session关闭后再访问user.addresses会直接抛DetachedInstanceError。解决方式要么在Session内完成数据组装要么用joinedload等预加载策略把关联数据一并拉回来要么用expire_on_commitFalse并保持对象在会话内可访问。4. 修改与删除先查后改与批量操作4.1 先查后改是最推荐的更新方式修改数据我永远推荐“先查后改”而不是直接写UPDATE原因有三个能校验数据是否存在、能触发ORM层面的属性变更追踪、代码可读性最好。user session.get(User, 1) if user is None: raise ValueError(用户不存在) user.username new_name user.email new_emailexample.com session.commit()session.get(Model, pk)是按主键查询最直接的方式内部优先从Session缓存里找命中就不发SQL了。字段改动后commit时SQLAlchemy会自动比较旧值和新值只把发生变化的列更新进数据库其余列不动。4.2 批量更新的高效写法如果你要同时更新几千条记录的状态逐条先查后改会产生大量SELECT和UPDATE性能很差。这时候适合用Query.update直接发一条UPDATE语句from sqlalchemy import update # 写法一query.update() session.query(User).filter(User.status inactive).update( {status: disabled}, synchronize_sessionFalse ) # 写法二update()核心语句 session.execute( update(User) .where(User.status inactive) .values(statusdisabled) ) session.commit()这两种写法有什么差别session.query(...).update()是ORM风格的批量更新可以指定synchronize_session参数来控制是否同步本地缓存中的对象状态sqlalchemy.update()则更接近核心表达式风格。我日常使用前者更多因为它和Query链式查询风格更一致。注意批量更新默认不会改动已加载在Session里的Python对象属性如果你接下来还要用这些对象最好重新查询或显式同步。4.3 删除单条与批量删除删除单条数据同样还是先取对象再调deleteuser session.get(User, 1) if user: session.delete(user) session.commit()批量删除用Query.delete就能一条DELETE语句解决问题session.query(User).filter(User.status disabled).delete(synchronize_sessionFalse) session.commit()synchronize_sessionFalse在这里的含义很重要告诉ORM不用挨个同步Session里已加载对象的状态速度会快很多。但代价是这个Session里如果还有这些被删对象它们不会自动变成“已删除”后续访问可能出现状态不一致所以用完这个Session后尽量别继续操作相关对象。4.4 物理删除与软删除的取舍在业务系统里“删除”不等于“销毁”。我很建议你在模型层直接设计软删除字段比如is_deleted或deleted_at因为物理删除的审计成本很高一旦误删很难恢复。物理删除也不是不能用比如临时缓存表、流水清算表这种数据量大且不需要追溯的业务物理删除反而干净。软删除的查询要处处带filter(is_deletedFalse)非常容易漏所以在模型层把默认过滤条件做进去会更稳妥。SQLAlchemy 2.x里可以通过__mapper_args__的with_polymorphic或者自定义Query来实现也可以在基类里统一封装class BaseMixin: classmethod def query_active(cls, session): if hasattr(cls, is_deleted): return session.query(cls).filter(cls.is_deleted False) # noqa: E712 return session.query(cls)4.5 级联删除的注意点当你删除一个父对象时关联的子对象要不要跟着删这取决于外键约束和relationship里的cascade配置。默认情况下session.delete会尝试把关联对象的状态同步删除但如果数据库外键设置了RESTRICTORM层的删除可能会在commit时报外键约束错误。我的建议是外键约束和ORM的cascade策略保持一套别两边各写一套。比如希望在删除用户时同时删除其地址记录数据库外键用ON DELETE CASCADEORM里relationship设置cascadeall, delete-orphan两者配合行为才最一致。否则很容易出现“ORM认为删了数据库却报错”的诡异情况。5. 事务、回滚与常见问题排查5.1 Session生命周期与异常回滚Session既是操作入口也管理着事务。session.commit()成功代表事务提交出现异常时必须session.rollback()回滚否则Session会卡在一个不一致的状态连接也无法正常归还连接池。最省心的方式是配合上下文管理器with SessionLocal() as session: try: session.add(new_user) session.commit() except Exception: session.rollback() raise这里的SessionLocal()上下文管理器只在退出时负责关闭Session不负责自动提交或回滚所以事务控制还是要自己写。我见过不少同事以为离开with就是自动commit结果数据半天没落库排查好久才找到原因。5.2 连接池耗尽问题在Web应用里Session用完后没有关闭会导致连接一直不归还连接池最终报出“Cant connect to MySQL server”或连接池超时。这个现象并不是数据库挂了而是连接泄漏。关键经验Session的创建和关闭要遵循“请求级”生命周期不要在全局创建一个大Session到处用也不要每次操作都创建一个Session。我比较推荐在Web框架的依赖注入机制里注册Session生成器请求开始创建一个Session请求结束统一关闭配合contextvars或作用域会话既安全又高效。会话执行完数据库操作后即使不调用close只要超出引用范围Session也会被垃圾回收并释放连接。但这是不可控的高并发下很容易出问题。显式关闭永远比依赖GC更稳。5.3 常见报错与解决速查表报错信息原因解决方案DetachedInstanceError在Session关闭后访问未预加载的关联属性用joinedload预加载或在会话内完成数据组装PendingRollbackError事务异常后未回滚又继续操作异常块中先rollback再处理InvalidRequestError: Object is already attached to session同一对象被add到多个Session保证对象在一个Session内流转IntegrityError唯一约束冲突或外键完整性失败用savepoint或事务包裹捕获异常单独处理OperationalError: (pymysql.err.OperationalError) ... Lost connection连接空闲过久或被kill使用连接池预检设置合适的pool_recycleUnmappedInstanceError传入ORM的类并没有被mapper映射检查模型类是否继承了DeclarativeBase这里特别说一下Lost connection。MySQL默认的wait_timeout一般是8小时但业务空闲几分钟就可能断开如果连接池没有及时回收失效连接一旦复用就可能报错。因此创建engine时我习惯设置pool_pre_pingTrue每次从连接池取连接时先ping一下失效连接直接丢掉重连虽然有一点点性能开销但能避免大量诡异报错。5.4 实际项目中的几点经验第一个经验是关于分页和排序的组合。列表页的排序字段要尽量带唯一字段比如id作为次级排序否则MySQL在相同排序值之间不保证顺序稳定翻页容易重复或遗漏。第二个经验是批量操作后刷新数据。用了bulk_insert_mappings或query.update()、query.delete()之后如果Session里还缓存着旧数据后面读取可能拿到过期值。稳妥做法是把当前Session的expire_all()调一下强制下次访问属性时重新查库避免数据不一致。第三个经验是谨慎使用__all__。:.all()会把所有结果加载进内存如果结果集有几万行内存占用会直接飙升。批量数据处理时用yield_per方法配合流式查询每次只取一小批数据到内存效率会好很多。例如for user in session.query(User).yield_per(100): process(user)第四个经验事务的粒度一定要细。我见过把大量业务逻辑塞进一个大事务里中间某个步骤失败导致整个事务回滚前面所有操作全部白费。更合理的做法是拆成多个小事务每个事务只负责一组内聚的操作失败时只回滚对应事务块并记录业务日志用于补偿。结尾从跑通到跑稳增删改查看起来简单但真正写得稳、跑得快需要理解Session、事务、加载策略这些底层机制。我个人最大的体会是不要急着写复杂代码先把单个Session里add、flush、commit、rollback的时序理清楚很多莫名其妙的报错都能迎刃而解。建议你把本文里的代码按顺序跑一遍再尝试混合使用条件查询和批量更新遇到问题对照第5节的速查表排查一次基本就能完成从“会用”到“用得稳”的过渡。
前两篇把SQLAlchemy的环境配置、模型映射和连接池管理都过了一遍这篇直接进入正题——用SQLAlchemy完成增删改查CRUD。这是所有业务系统里最基础、最高频的操作也是很多人入门ORM后第一道必须跨过的门槛。我尽量把每个操作背后的逻辑讲透比如为什么推荐先查再改、flush和commit到底差在哪、批量操作怎么写才优雅最后附上我实际跑项目时踩过的一些坑。1. 增删改查的整体思路与操作前置准备1.1 为什么单独把CRUD拎出来讲很多教程把增删改查分散在环境搭建和模型定义里看起来省篇幅实际上对新手很不友好。CRUD并不是简单的“换成SQL语法写一遍”里面有很多ORM特有的机制比如会话状态、主键回填、懒加载触发时机这些不彻底搞清楚写出来的代码往往“能用但味道不对”后期并发一上来就各种报错。我在参与某个Web系统的数据层重构时看到过一段让我印象很深的代码——每次查询都手动创建一个新连接用完也不关数据库连接池直接被拖垮。问题不在技术难度而是对Session的生命周期没有概念。所以这篇会把Session机制作为开篇重点它是理解整个数据库操作入口的钥匙然后逐项拆解增、删、改、查。1.2 一切操作都从Session开始SQLAlchemy 2.x 里ORM操作的核心对象是Session。可以把它理解成一次“工作单元”的载体负责跟踪模型对象的变更状态并在合适的时机把变更同步回数据库。一个Session内部持有数据库连接但它并不等同于连接更准确地说它是在连接之上封装了一层对象状态管理。from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, Session engine create_engine( mysqlpymysql://username:passwordlocalhost:3306/demo?charsetutf8mb4, echoFalse ) SessionLocal sessionmaker(bindengine, autoflushFalse, expire_on_commitFalse)这里有两个参数值得展开。autoflushFalse意味着查询前不会自动把未提交的修改先刷到数据库这在后面讲“先查后改”时至关重要expire_on_commitFalse表示提交后对象上的属性不会被强制过期下次访问属性时不用再发一次SQL对性能敏感的业务很友好。实际项目中我习惯把SessionLocal做成模块级单例再用依赖注入或上下文管理器传给业务函数。1.3 ORM操作和原生SQL的映射关系初学者最容易懵的一点明明写的是Python对象操作怎么就到数据库里变成了一条SQL其实每一步操作都有对应的SQL语义。ORM操作底层SQL语义关键方法/属性新增对象INSERT INTO ...session.add(obj)/session.commit()按主键查询SELECT ... WHERE id ?session.get(Model, pk)条件筛选SELECT ... WHERE ...session.query(Model).filter(...)修改对象属性UPDATE ... SET ...先查后改 /session.query(Model).update(...)删除对象DELETE FROM ...session.delete(obj)/query.delete()搞清楚这个映射关系后写CRUD会有一种“心里有底”的感觉。接下来的每一节我都按这个表格展开配合可运行代码和补充参数说明来讲。2. 新增数据从单条到批量的完整姿势2.1 单条新增最简写法假设已经定义好一个User模型包含username、email、created_at三个字段。新增一条用户记录最常规的写法如下with SessionLocal() as session: new_user User( usernamealice, emailaliceexample.com, created_atdatetime.now() ) session.add(new_user) session.commit()有人可能会问add之后不commit行不行不执行commit事务就不会提交数据不会真正写入数据库Session关闭时未提交的修改会被回滚丢掉。commit的行为类似于保存文件——add只是放在了“内存草稿箱”commit才是“点击保存”。2.2 新增后立刻拿到自增主键很多业务场景需要新增后马上用主键做一些后续操作。SQLAlchemy在commit之后会自动把数据库返回的自增ID回填到对象的主键属性上with SessionLocal() as session: new_user User(usernamebob, emailbobexample.com) session.add(new_user) session.commit() user_id new_user.id # 这里已经是数据库里真实的主键值在这个回填机制中其实起作用的是flush。如果等不到commit也可以手动flush来提前触发主键回填——但flush只是把SQL发给数据库事务仍未提交后续出错还可以整体回滚。这也是我比较推荐的一种方式尤其适合需要先拿到ID再关联写入子表数据的场景。session.add(new_user) session.flush() user_id new_user.id # flush后主键已回填2.3 批量新增的两个常用方案一次插入几十上百条数据时逐条add再commit显然太低效。常见方案有两种。第一种是用add_all把对象列表一次性加入会话users [ User(usernamefuser_{i}, emailfuser_{i}example.com) for i in range(100) ] session.add_all(users) session.commit()第二种是直接用session.bulk_insert_mappings绕开完整的ORM状态跟踪速度更快但不会自动回填主键也不会触发模型上的事件监听session.bulk_insert_mappings( User, [ {username: fuser_{i}, email: fuser_{i}example.com} for i in range(100) ] ) session.commit()至于哪种更好我的意见是几十条的量级两者差距不明显优先add_all语义更统一超过几百条再做批量插入时可以考虑bulk_insert_mappings能明显加快整体插入速度。如果再往上走比如一次性导入上万条就该考虑executemany或COPY类方案了。2.4 实操心得主键回填失败的排查方向我在某次批量导入过程中遇到过主键回填失效排查半天发现是模型里主键字段没有设置为autoincrement。MySQL默认对整数主键开启自增但如果你显式定义了primary_keyTrue却没有挂autoincrement有些数据库方言下回填会出问题。经验法则只要主键是单一整数直接写mapped_column(Integer, primary_keyTrue, autoincrementTrue)省心很多。另外如果使用bulk_insert_mappings也别指望它能回填主键需要回填ID时请绕回add/add_all。3. 查询操作filter、条件组合与结果处理查询是CRUD里占比最高的部分也是优化收益最明显的部分。我按“基础查询→条件组合→排序分页→聚合关联”四个层次来讲。这一节内容比较长建议配合代码实际操作一遍。3.1 query.filter与filter_by的区别query(Model)是Query对象构建的起点然后可以在它后面链条式追加过滤条件。两种过滤方法# filter: 接收SQLAlchemy表达式 session.query(User).filter(User.id 10).all() # filter_by: 接收关键字参数只支持等值匹配 session.query(User).filter_by(usernamealice).first()filter的表达能力更强各种、、like、in、or_都靠它filter_by本质是filter(Model.attr value)的简写适合简单等值场景。不要混用太狠保持同一种风格代码读起来更整洁。3.2 常用查询条件速查表需求写法等值匹配User.id 10不等于User.id ! 10模糊匹配User.username.like(%alice%)范围匹配User.age.between(18, 30)包含于集合User.id.in_([1, 2, 3])为空判断User.email.is_(None)/isnot(None)多个条件同时满足(AND)and_(User.id 1, User.age 30)多个条件满足其一(OR)or_(User.id 1, User.username admin)实际使用时我建议优先用and_、or_显式组合可读性更好。也可以直接用Python的和|运算符但要给每个条件加括号否则容易踩运算符优先级的坑例如from sqlalchemy import and_, or_ condition and_( User.age 18, or_(User.username.like(alice%), User.email.like(%example.com)) ) results session.query(User).filter(condition).all()3.3 排序、分页、限量查询排序用order_by支持多个字段组合。分页有两种惯用方式经典limit/offset和按主键游标分页。对于后台管理系统中小数据量的分页limit/offset完全够用一旦数据量过十万offset偏移越大越慢因为数据库要扫描和丢弃前面的行。# 排序 限量 session.query(User).order_by(User.created_at.desc(), User.id.asc()).limit(10).all() # 分页第2页每页20条 page, per_page 2, 20 session.query(User).order_by(User.id).offset((page - 1) * per_page).limit(per_page).all() # 游标分页以last_id为游标取之后20条 session.query(User).filter(User.id last_id).order_by(User.id).limit(per_page).all()limit(0)在SQLAlchemy里会被忽略做“取全部”时要注意这个细节。另外order_by尽量跟随索引走否则分页查询数据量一大很容易拖垮数据库性能。3.4 只查需要的字段避免整行加载不需要全部列时用with_entities或直接用旧式query(Model.字段名)限制返回列这样既省内存又加快查询。我见过不少同事习惯性query(Order).all()然后只取其中两个字段数据量大时白白浪费很多IO。# 方案一查询指定列 rows session.query(User.id, User.username).filter(User.age 18).all() # 方案二查询完整实体但只加载部分字段 rows session.query(User).options(load_only(User.id, User.username)).all()如果你只是想判断“是否存在”正确做法是用session.query(User.id).filter(...).first()而不是查全表然后判断list是否为空。判断存在时也可以直接用exists()from sqlalchemy import exists exists_flag session.query( exists().where(User.email aliceexample.com) ).scalar()3.5 聚合与分组聚合函数是常用能力。统计用户数、按状态分组统计数量、求最大最小值都集中在func里from sqlalchemy import func # 总数 total session.query(func.count(User.id)).scalar() # 按状态分组统计 rows session.query(User.status, func.count(User.id)).group_by(User.status).all() # 数量大于5的状态分组 rows ( session.query(User.status, func.count(User.id).label(cnt)) .group_by(User.status) .having(func.count(User.id) 5) .all() )注意scalar()只取第一行第一列的值聚合查询返回单值时非常方便。分组统计如果要过滤分组一定要用having放到where里会报错或者得到错误语义。3.6 关联查询与加载策略实际项目很少只有单表查询。User和Address、Order和OrderItem这类一对多、多对一关系非常普遍。SQLAlchemy关联查询的用法# 通过relationship属性访问关联对象默认懒加载 user session.query(User).filter(User.id 1).first() addresses user.addresses # 此时才触发额外的SELECT # 显式join查询 rows ( session.query(User, Address) .join(Address, User.id Address.user_id) .filter(Address.city Shanghai) .all() ) # 预先加载关联对象避免N1查询 from sqlalchemy.orm import joinedload users ( session.query(User) .options(joinedload(User.addresses)) .all() )懒加载是很多人容易踩坑的重灾区。在Session关闭后再访问user.addresses会直接抛DetachedInstanceError。解决方式要么在Session内完成数据组装要么用joinedload等预加载策略把关联数据一并拉回来要么用expire_on_commitFalse并保持对象在会话内可访问。4. 修改与删除先查后改与批量操作4.1 先查后改是最推荐的更新方式修改数据我永远推荐“先查后改”而不是直接写UPDATE原因有三个能校验数据是否存在、能触发ORM层面的属性变更追踪、代码可读性最好。user session.get(User, 1) if user is None: raise ValueError(用户不存在) user.username new_name user.email new_emailexample.com session.commit()session.get(Model, pk)是按主键查询最直接的方式内部优先从Session缓存里找命中就不发SQL了。字段改动后commit时SQLAlchemy会自动比较旧值和新值只把发生变化的列更新进数据库其余列不动。4.2 批量更新的高效写法如果你要同时更新几千条记录的状态逐条先查后改会产生大量SELECT和UPDATE性能很差。这时候适合用Query.update直接发一条UPDATE语句from sqlalchemy import update # 写法一query.update() session.query(User).filter(User.status inactive).update( {status: disabled}, synchronize_sessionFalse ) # 写法二update()核心语句 session.execute( update(User) .where(User.status inactive) .values(statusdisabled) ) session.commit()这两种写法有什么差别session.query(...).update()是ORM风格的批量更新可以指定synchronize_session参数来控制是否同步本地缓存中的对象状态sqlalchemy.update()则更接近核心表达式风格。我日常使用前者更多因为它和Query链式查询风格更一致。注意批量更新默认不会改动已加载在Session里的Python对象属性如果你接下来还要用这些对象最好重新查询或显式同步。4.3 删除单条与批量删除删除单条数据同样还是先取对象再调deleteuser session.get(User, 1) if user: session.delete(user) session.commit()批量删除用Query.delete就能一条DELETE语句解决问题session.query(User).filter(User.status disabled).delete(synchronize_sessionFalse) session.commit()synchronize_sessionFalse在这里的含义很重要告诉ORM不用挨个同步Session里已加载对象的状态速度会快很多。但代价是这个Session里如果还有这些被删对象它们不会自动变成“已删除”后续访问可能出现状态不一致所以用完这个Session后尽量别继续操作相关对象。4.4 物理删除与软删除的取舍在业务系统里“删除”不等于“销毁”。我很建议你在模型层直接设计软删除字段比如is_deleted或deleted_at因为物理删除的审计成本很高一旦误删很难恢复。物理删除也不是不能用比如临时缓存表、流水清算表这种数据量大且不需要追溯的业务物理删除反而干净。软删除的查询要处处带filter(is_deletedFalse)非常容易漏所以在模型层把默认过滤条件做进去会更稳妥。SQLAlchemy 2.x里可以通过__mapper_args__的with_polymorphic或者自定义Query来实现也可以在基类里统一封装class BaseMixin: classmethod def query_active(cls, session): if hasattr(cls, is_deleted): return session.query(cls).filter(cls.is_deleted False) # noqa: E712 return session.query(cls)4.5 级联删除的注意点当你删除一个父对象时关联的子对象要不要跟着删这取决于外键约束和relationship里的cascade配置。默认情况下session.delete会尝试把关联对象的状态同步删除但如果数据库外键设置了RESTRICTORM层的删除可能会在commit时报外键约束错误。我的建议是外键约束和ORM的cascade策略保持一套别两边各写一套。比如希望在删除用户时同时删除其地址记录数据库外键用ON DELETE CASCADEORM里relationship设置cascadeall, delete-orphan两者配合行为才最一致。否则很容易出现“ORM认为删了数据库却报错”的诡异情况。5. 事务、回滚与常见问题排查5.1 Session生命周期与异常回滚Session既是操作入口也管理着事务。session.commit()成功代表事务提交出现异常时必须session.rollback()回滚否则Session会卡在一个不一致的状态连接也无法正常归还连接池。最省心的方式是配合上下文管理器with SessionLocal() as session: try: session.add(new_user) session.commit() except Exception: session.rollback() raise这里的SessionLocal()上下文管理器只在退出时负责关闭Session不负责自动提交或回滚所以事务控制还是要自己写。我见过不少同事以为离开with就是自动commit结果数据半天没落库排查好久才找到原因。5.2 连接池耗尽问题在Web应用里Session用完后没有关闭会导致连接一直不归还连接池最终报出“Cant connect to MySQL server”或连接池超时。这个现象并不是数据库挂了而是连接泄漏。关键经验Session的创建和关闭要遵循“请求级”生命周期不要在全局创建一个大Session到处用也不要每次操作都创建一个Session。我比较推荐在Web框架的依赖注入机制里注册Session生成器请求开始创建一个Session请求结束统一关闭配合contextvars或作用域会话既安全又高效。会话执行完数据库操作后即使不调用close只要超出引用范围Session也会被垃圾回收并释放连接。但这是不可控的高并发下很容易出问题。显式关闭永远比依赖GC更稳。5.3 常见报错与解决速查表报错信息原因解决方案DetachedInstanceError在Session关闭后访问未预加载的关联属性用joinedload预加载或在会话内完成数据组装PendingRollbackError事务异常后未回滚又继续操作异常块中先rollback再处理InvalidRequestError: Object is already attached to session同一对象被add到多个Session保证对象在一个Session内流转IntegrityError唯一约束冲突或外键完整性失败用savepoint或事务包裹捕获异常单独处理OperationalError: (pymysql.err.OperationalError) ... Lost connection连接空闲过久或被kill使用连接池预检设置合适的pool_recycleUnmappedInstanceError传入ORM的类并没有被mapper映射检查模型类是否继承了DeclarativeBase这里特别说一下Lost connection。MySQL默认的wait_timeout一般是8小时但业务空闲几分钟就可能断开如果连接池没有及时回收失效连接一旦复用就可能报错。因此创建engine时我习惯设置pool_pre_pingTrue每次从连接池取连接时先ping一下失效连接直接丢掉重连虽然有一点点性能开销但能避免大量诡异报错。5.4 实际项目中的几点经验第一个经验是关于分页和排序的组合。列表页的排序字段要尽量带唯一字段比如id作为次级排序否则MySQL在相同排序值之间不保证顺序稳定翻页容易重复或遗漏。第二个经验是批量操作后刷新数据。用了bulk_insert_mappings或query.update()、query.delete()之后如果Session里还缓存着旧数据后面读取可能拿到过期值。稳妥做法是把当前Session的expire_all()调一下强制下次访问属性时重新查库避免数据不一致。第三个经验是谨慎使用__all__。:.all()会把所有结果加载进内存如果结果集有几万行内存占用会直接飙升。批量数据处理时用yield_per方法配合流式查询每次只取一小批数据到内存效率会好很多。例如for user in session.query(User).yield_per(100): process(user)第四个经验事务的粒度一定要细。我见过把大量业务逻辑塞进一个大事务里中间某个步骤失败导致整个事务回滚前面所有操作全部白费。更合理的做法是拆成多个小事务每个事务只负责一组内聚的操作失败时只回滚对应事务块并记录业务日志用于补偿。结尾从跑通到跑稳增删改查看起来简单但真正写得稳、跑得快需要理解Session、事务、加载策略这些底层机制。我个人最大的体会是不要急着写复杂代码先把单个Session里add、flush、commit、rollback的时序理清楚很多莫名其妙的报错都能迎刃而解。建议你把本文里的代码按顺序跑一遍再尝试混合使用条件查询和批量更新遇到问题对照第5节的速查表排查一次基本就能完成从“会用”到“用得稳”的过渡。
平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测)
