Python SQLAlchemy ORM 框架从入门到精通：核心概念与CRUD实战指南-北京尧图网络

什么是SQLAlchemySQLAlchemy是Python中最流行的ORM对象关系映射框架它让开发者可以用Python对象操作数据库无需直接编写SQL语句。核心优势跨数据库兼容支持MySQL、PostgreSQL、SQLite等灵活的查询方式强大的事务支持完善的ORM映射机制SQLAlchemy核心概念SQLAlchemy分为两大模块Core核心组件和ORM对象关系映射。Core核心组件Engine引擎管理数据库连接Session会话数据库操作的桥梁Base基类所有ORM模型类的父类Model模型Python类对应数据库表Column字段定义表的字段和约束ORM对象关系映射ORM将Python对象与数据库表进行映射。核心优势无需编写原生SQL降低学习成本代码与数据库解耦便于维护和迁移自动处理数据类型转换支持事务管理和关联查询快速开始环境准备1. 安装SQLAlchemy及MySQL驱动pipinstallsqlalchemy pymysql2. 创建数据库CREATEDATABASEsqlalchemy_demo;创建Engine连接数据库fromsqlalchemyimportcreate_engine# 连接字符串格式数据库驱动://用户名:密码主机:端口/数据库名db_urlmysqlpymysql://root:123456localhost:3306/sqlalchemy_demoenginecreate_engine(db_url,echoTrue)# echoTrue打印SQL方便调试创建Base基类和Model模型fromsqlalchemy.ext.declarativeimportdeclarative_basefromsqlalchemyimportColumn,Integer,String,DateTimefromdatetimeimportdatetime# 创建Base基类Basedeclarative_base()# 定义User模型类classUser(Base):__tablename__user# 表名idColumn(Integer,primary_keyTrue,autoincrementTrue)usernameColumn(String(50),nullableFalse,uniqueTrue)passwordColumn(String(100),nullableFalse)create_timeColumn(DateTime,defaultdatetime.now)def__repr__(self):returnfUser(id{self.id}, username{self.username})创建数据表# 根据模型类自动创建表Base.metadata.create_all(engine)print(数据库表创建成功)创建Session实现数据库操作fromsqlalchemy.ormimportsessionmaker# 创建Session工厂SessionFactorysessionmaker(bindengine)# 插入数据sessionSessionFactory()userUser(usernamezhangsan,password123456)session.add(user)session.commit()session.close()# 查询数据sessionSessionFactory()all_userssession.query(User).all()foruinall_users:print(u)session.close()WHERE username ! ‘zhangsan’user2 session.query(User).filter(User.username ! “zhangsan”).all()print(f’{user2}‘)# 3. 模糊查询like对应 MySQL 的 WHERE username LIKE ‘%a%’user3 session.query(User).filter(User.username.like(“%a%”)).all()print(f’{user3}‘)# 4. 范围查询in_对应 MySQL 的 WHERE id IN (1,2,3)user4 session.query(User).filter(User.id.in_([1, 2, 3])).all()print(f’{user4}‘)# 5. 大于、小于、大于等于、小于等于user5 session.query(User).filter(User.id 2).all() # 对应 WHERE id 2print(f’{user5})# 6. 多条件查询and_ / or_对应 MySQL 的 AND / ORfrom sqlalchemy import and_, or_# 对应 WHERE id 1 AND username LIKE w% user6 session.query(User).filter(and_(User.id 1, User.username.like(w%))).first() print(f{user6}) # 对应 WHERE usernamezhangsan AND password123456 user7 session.query(User).filter_by(usernamezhangsan, password123456).first() print(f{user7}) # 对应 WHERE id 1 OR username LIKE w% user8 session.query(User).filter(or_(User.id 1, User.username.like(w%))).all() print(f{user8})#### 排序、分页查询 python from sqlalchemy_config import SessionFactory from base_model import User #导入Model模型 with SessionFactory() as session: # 1. 排序order_by对应 MySQL 的 ORDER BY users_asc session.query(User).order_by(User.id).all() # 按 id 升序默认对应 ORDER BY id ASC users_desc session.query(User).order_by(User.id.desc()).all() # 按 id 降序对应 ORDER BY id DESC print(f{users_asc}) print(f{users_desc}) # 2. 分页limit offset对应 MySQL 的 LIMIT 和 OFFSET适合分页查询 page 2 # 第 2 页 page_size 2 # 每页 2 条数据 # 对应 SELECT * FROM user ORDER BY id LIMIT 2 OFFSET 2 users_page session.query(User).order_by(User.id).limit(page_size).offset((page - 1) * page_size).all() print(f{users_page}) # 3. 统计数量count对应 MySQL 的 COUNT(*) user_count session.query(User).filter(User.username.like(z%)).count() print(f{user_count})分组查询函数作用func.count(字段)统计数量func.sum(字段)求和func.avg(字段)平均值func.max(字段)最大值func.min(字段)最小值fromsqlalchemy_configimportsessionFactory,funcfromsqlalchemyimportselectfrombase_modelimportUser,Accountwith(sessionFactory()assession):# 1.简单分组查询# 因为聚合函数是数据库提供的函数所以需要导入func# 默认query(Account)查询所有列聚合查询只能查询分组列与聚合函数所以# 所以需要查询指定列group1session.query(Account.age,func.count(Account.id)).group_by(Account.age).all()print(group1)# 聚合比较特殊不会返回对应的对象所以默认以元组返回# 如果想以字典形式返回需要使用新版2.0 API然后使用.mappings()stmtselect(Account.age,func.count(Account.id).label(count)).group_by(Account.age)group2session.execute(stmt).mappings().all()print(group2)# 2.having分组筛选# 可以直接使用having()进行分组的筛选group3session.query(Account.age,func.count(Account.id)).group_by(Account.age).having(func.count(Account.id)1).all()print(group3)stmtselect(Account.age,func.count(Account.id).label(count)).group_by(Account.age).having(func.count(Account.id)1)group4session.execute(stmt).mappings().all()print(group4)Create新增数据新增数据的步骤创建模型实例 → 将实例添加到会话 → 提交会话commit新增后的数据会直接写入 MySQL 数据库。fromsqlalchemy_configimportSessionFactoryfrombase_modelimportUser#导入Model模型withSessionFactory()assession:# 创建User实例相当于创建一条数据user1User(usernamelisi2,password123456)user2User(usernamewangwu2,password654321)# 1. 将实例添加到Session相当于“暂存”数据未提交到数据库session.add(user1)session.add(user2)# 提交事务将暂存的数据提交到数据库此时才会真正插入数据session.commit()# 2. 批量添加可选session.add_all([user1,user2])session.commit()# 3. 批量字典添加# 准备字典列表不用创建User对象user_dicts[{username:sunqi,password:123123},{username:zhouba,password:456456},{username:wujiu,password:789789},]# 4. 批量插入session.bulk_insert_mappings(User,user_dicts)session.commit()# 5. sqlalchemy2.0语法fromsqlalchemyimportinsert# 构建insert语句stmtinsert(User).values([{username:zhengshi,password:aaa},{username:chenshi,password:bbb},])# 执行语句session.execute(stmt)session.commit()Delete删除数据删除数据的步骤查询数据 → 删除实例 → 提交会话删除操作会直接删除 MySQL 中的对应数据。fromsqlalchemy_configimportSessionFactoryfrombase_modelimportUser#导入Model模型fromsqlalchemyimportdeletewithSessionFactory()assession:# 1. 单条数据删除usersession.query(User).filter(User.usernamewangwu).first()ifuser:session.delete(user)# 对应 DELETE FROM user WHERE username wangwusession.commit()# 2. 批量删除对应 MySQL 的 DELETE ... WHERE ...# 删除所有 id 3 的用户session.query(User).filter(User.id3).delete()session.commit()# 3. sqlalchemy2.0语法fromsqlalchemyimportdelete# 构建 delete 语句stmtdelete(User).where(User.id18)# 执行语句session.execute(stmt)session.commit()Update更新数据更新数据的步骤查询数据 → 修改实例属性 → 提交会话更新操作会直接同步到 MySQL 数据库。fromsqlalchemy_configimportSessionFactoryfrombase_modelimportUser#导入Model模型withSessionFactory()assession:# 1. 单条数据更新usersession.query(User).filter(User.usernamezhangsan).first()ifuser:user.passwordnew_password123# 修改密码对应 UPDATE user SET password new_password123session.commit()# 提交更新print(f{user})# 2. 批量更新高效对应 MySQL 的 UPDATE ... WHERE ...# 将所有用户名以 w 开头的用户密码更新为 common_passwordsession.query(User).filter(User.username.like(w%)).update({password:common_password})session.commit()# 3. sqlalchemy2.0语法fromsqlalchemyimportupdate# 构建 update 语句stmtupdate(User).where(User.username.like(w%)).values(passwordsqlalchemy2.0_password)# 执行语句session.execute(stmt)session.commit()

Python SQLAlchemy ORM 框架从入门到精通：核心概念与CRUD实战指南

相关推荐

Samba安全加固：修改默认端口与Windows客户端连接配置详解

LeetCode 1888题解析：二进制字符串交替转换的最优解法

5V转1.2V电源设计：LDO与开关降压方案深度对比与实战选型

更多资讯

Node-Red环境搭建避坑指南：从零跑通第一个节点

Context-Mode：一种解决代码导航迷航的上下文模式

Aurelio 上手：一个本地优先的个人财富管理应用，AI 只提建议不直改数据

Litellm实战：统一多LLM API网关的部署与排坑指南

基于Spark与Django的胆结石数据分析与可视化系统设计

lsp-mode 故障排查完全指南：从日志、服务器路径到最小环境复现

今日更新

声子晶体仿真：三角晶格与六角晶格的能带计算实战

度假村安保定岗定编：岗位梳理、编制测算与排班设计

COMSOL等离子体模块实战：氩气双层DBD仿真建模全解析

本周精选

交通拥堵预测毕设：基于Python+LSTM的车辆流量预测系统全流程解析

MATLAB/Simulink搭建10机39节点电力系统暂态稳定仿真实战

无人机航拍三维重建全流程：从SfM到网格生成的避坑指南

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？