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()