前言SQLAlchemy是 Python 中最流行的数据库工具库分为两大核心部分ORM对象关系映射 Object Relational Mapper【最常用】Core底层 SQL 表达式工具什么是ORM一种编程技术用于实现面向对象编程语言中的对象与关系型数据库中的表之间的映射。数据库SQLAlchemy ORM表 (table)模型类继承Base字段 (column)类属性Column()一条记录对象实例SELECT/INSERT/UPDATE/DELETE对象方法调用什么是Core他是SQL 语句构造器表达式语言是 SQLAlchemy 的底层基础不依赖 ORM、不需要定义模型类实现用代码动态组装标准 SQL自动处理参数防止 SQL 注入。思想直接构造表、字段、SQL 语句只定义表结构Table没有对象映射操作对象表、列、查询表达式返回原始行数据元组 / 字典SQL Alchemy重要组件Engine引擎SQLAlchemy的核心入口负责管理数据库连接通过连接字符串创建决定了数据库类型、地址、账号密码等信息。Session会话用于与数据库进行交互的“桥梁”所有数据库操作增删改查都通过Session完成相当于一个临时的数据库连接会话。Base基类所有ORM模型类的父类通过Base类创建的子类会自动映射为数据库中的表。Model模型Python中的类对应数据库中的一张表类的属性对应表的字段类的实例对应表的一行数据。Column字段用于定义模型类的属性即数据库表的字段可指定字段类型、主键、非空、默认值等约束基础实现0、环境准备安装相应的库首先安装SQLAlchemy及对应数据库的驱动以MySQL为例# 安装SQLAlchemy核心库 pip install sqlalchemy # 安装MySQL驱动常用两种二选一 pip install pymysql # 纯Python实现兼容性好 pip install mysql-connector-python # MySQL官方驱动数据库准备在开始使用 SQLAlchemy 操作 MySQL 前需提前准备好 MySQL 数据库后续 SQLAlchemy 仅操作表方法一使用图形化界面Nvicate创建数据库输入数据库名称以及密码方法二命令行窗口登录mysql命令# 1.登录 MySQL 数据库 mysql -u 用户名 -p创建数据库命令# 2.创建数据库示例数据库名sqlalchemy_demo CREATE DATABASE IF NOT EXISTS sqlalchemy_demo CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; # 3.确认数据库创建成功 SHOW DATABASES; # 若列表中出现 sqlalchemy_demo则数据库准备完成。TIPSsql代码需要通过分号结尾不然判断不到语句是否结束1、创建Engine连接数据库这一步实现的主要功能就是连接到刚刚创建的数据库表from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine # 1. 定义MySQL连接字符串格式数据库驱动://用户名:密码主机:端口/数据库名?参数 # 以pymysql驱动为例 db_url mysqlpymysql://root:123456localhost:3306/sqlalchemy_demo?charsetutf8mb4 # 2. 创建EngineechoTrue表示打印执行的SQL语句方便调试生产环境可关闭 engine create_engine(db_url, echoTrue)说明root数据库用户名替换为自己的数据库账号123456数据库密码替换为自己的密码sqlalchemy_demo要连接的数据库名需提前在MySQL中创建或后续通过代码创建echoTrue可选开启SQL打印调试时可清晰看到SQLAlchemy生成的原生SQL一般进行问题排查时添加。2、创建Base基类和Model模型通过Base基类创建模型类将数据库中的表映射到该模型类以“用户表user”为例from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String, DateTime from datetime import datetime # 1. 创建Base基类所有模型类必须继承此类 Base declarative_base() # 2. 定义User模型类对应数据库中的user表 class User(Base): # 定义表名如果不指定默认用类名小写作为表名 __tablename__ user # 定义字段id主键自增 id Column(Integer, primary_keyTrue, autoincrementTrue, comment用户ID) # 定义字段username用户名非空唯一 username Column(String(50), nullableFalse, uniqueTrue, comment用户名) # 定义字段password密码非空 password Column(String(100), nullableFalse, comment密码) # 定义字段create_time创建时间默认当前时间 create_time Column(DateTime, defaultdatetime.now, comment创建时间) # 可选定义__repr__方法方便打印实例时查看信息 def __repr__(self): return fUser(用户id{self.id}, 用户名{self.username}, 创建时间{self.create_time})其中Base为通用基类后续添加的所有模型类都必须继承Base模型类用于实现映射关系必须与数据库表中字段对应字段约束说明常用primary_keyTrue设为主键autoincrementTrue自增仅整数主键可用nullableFalse非空约束uniqueTrue唯一约束default默认值comment字段注释。定义__repr__方法的作用3、创建数据库表通过Base类的create_all()方法自动根据模型类创建数据库表如果表已存在则不会重复创建# 基于Base类创建所有模型对应的数据库表绑定Engine Base.metadata.create_all(engine) print(数据库表创建成功)CURD增删改查Session是与数据库交互的核心所有操作都需通过Session完成步骤为创建Session → 执行操作 → 提交事务 → 关闭Session0、抽取通用配置为了更清楚的进行SQL Alchemy的学习也为了养成良好的模块化编程的习惯所以将通用的配置部分代码进行抽取将配置抽象书写在sqlalchemy_config中统一定义与使用from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from sqlalchemy import create_engine # 1. 定义MySQL连接字符串格式数据库驱动://用户名:密码主机:端口/数据库名?参数 # 以pymysql驱动为例 db_url mysqlpymysql://root:123456localhost:3306/sqlalchemy_demo?charsetutf8mb4 # 2. 创建EngineechoTrue表示打印执行的SQL语句方便调试生产环境可关闭 engine create_engine(db_url, echoTrue) # 3. 创建Session工厂绑定Engine SessionFactory sessionmaker(bindengine) # 4. 创建Base基类所有模型类必须继承此类 Base declarative_base()模型类---base_modelfrom sqlalchemy import Column, Integer, String, DateTime from datetime import datetime from sqlalchemy_config import Base # 定义User模型类对应数据库中的user表 class User(Base): # 定义表名如果不指定默认用类名小写作为表名 __tablename__ user # 定义字段id主键自增 id Column(Integer, primary_keyTrue, autoincrementTrue, comment用户ID) # 定义字段username用户名非空唯一 username Column(String(50), nullableFalse, uniqueTrue, comment用户名) # 定义字段password密码非空 password Column(String(100), nullableFalse, comment密码) # 定义字段create_time创建时间默认当前时间 create_time Column(DateTime, defaultdatetime.now, comment创建时间) # 可选定义__repr__方法方便打印实例时查看信息 def __repr__(self): return fUser(用户id{self.id}, 用户名{self.username}, 创建时间{self.create_time})CRUD操作----main1、Read查询数据基础查询核心是query()函数from sqlalchemy_config import SessionFactory from base_model import User #导入Model模型 with SessionFactory() as session: print(1. 查询所有数据返回列表对应 MySQL 的 SELECT * FROM user) all_users session.query(User).all() for user in all_users: print(user) print(*40) print(2. 查询单条数据根据主键对应 MySQL 的 SELECT * FROM user WHERE id 1不存在则返回None) user_by_id session.get(User,2) # 1 是 id 值 print(f{user_by_id}) print( * 40) print(3. 查询单条数据不存在则返回 None对应 MySQL 的 SELECT * FROM user LIMIT 1) user_by_name session.query(User).first() print(f{user_by_name}) print( * 40) print(4. 查指定字段数据) # 默认.query(User)会查询User的所有字段 可以使用User.字段控制查询字段 all_users session.query(User.id,User.username,User.create_time).all() for user in all_users: print(user)输出结果条件查询核心是filter()函数添加查询条件支持 MySQL 所有常用运算符from sqlalchemy_config import SessionFactory from base_model import User #导入Model模型 with SessionFactory() as session: print(1. 等于对应 MySQL 的 WHERE username zhangsan) user1 session.query(User).filter(User.username zhangsan).first() print(f{user1}) print( * 40) print(2. 不等于!对应 MySQL 的 WHERE username ! zhangsan) user2 session.query(User).filter(User.username ! zhangsan).all() print(f{user2}) print( * 40) print(3. 模糊查询like对应 MySQL 的 WHERE username LIKE %a%) user3 session.query(User).filter(User.username.like(%a%)).all() print(f{user3}) print( * 40) print(4. 范围查询in_对应 MySQL 的 WHERE id IN (1,2,3)) user4 session.query(User).filter(User.id.in_([1, 2, 3])).all() print(f{user4}) print( * 40) print(5. 大于、小于、大于等于、小于等于) user5 session.query(User).filter(User.id 1).all() # 对应 WHERE id 2 print(f{user5}) print( * 40) print(6. 多条件查询and_ / or_对应 MySQL 的 AND / OR) from sqlalchemy import and_, or_ print(对应 WHERE id 1 AND username LIKE w%) user6 session.query(User).filter(and_(User.id 1, User.username.like(w%))).first() print(f{user6}) print( * 40) print(对应 WHERE usernamezhangsan AND password123456) user7 session.query(User).filter_by(usernamezhangsan, password123456).first() print(f{user7}) print( * 40) print(对应 WHERE id 1 OR username LIKE w%) user8 session.query(User).filter(or_(User.id 1, User.username.like(w%))).all() print(f{user8}) print( * 40)输出结果分组查询通过调用sqlalchemy.func中的聚合函数实现from sqlalchemy_config import SessionFactory from sqlalchemy import select, func from base_model import Account with (SessionFactory() as session): # 方法一简单分组查询 # 因为聚合函数是数据库提供的函数 所以需要导入func # 默认query(Account)查询所有列 聚合查询只能查询分组列与聚合函数 # 所以需要查询指定列 group1session.query(Account.age,func.count(Account.id)).group_by(Account.age).all() # 聚合比较特殊 不会返回对应的对象 默认以元组返回 print(group1) # 方法一的第二种实现 #如果想以字典形式返回需要使用新版2.0 API然后使用.mappings() stmt select( Account.age, func.count(Account.id).label(count) ).group_by(Account.age) group2 session.execute(stmt).mappings().all() print(group2) # 方法二having分组筛选 # 可以直接使用having()进行分组之后的筛选 group3 session.query(Account.age, func.count(Account.id)).group_by(Account.age).having(func.count(Account.id)1).all() print(group3) stmt select( Account.age, func.count(Account.id).label(count) ).group_by(Account.age).having(func.count(Account.id)1) group4 session.execute(stmt).mappings().all() print(group4)2、Create新增数据新增数据的步骤创建模型实例 → 将实例添加到会话 → 提交会话commit新增后的数据会直接写入 MySQL 数据库。from sqlalchemy_config import SessionFactory from base_model import User #导入Model模型 with SessionFactory() as session: # 创建User实例相当于创建一条数据 user1 User(usernamelisi2, password123456) user2 User(usernamewangwu2, password654321) # 1. 将实例添加到Session相当于“暂存”数据未提交到数据库 session.add(user1) session.add(user2) # 提交事务将暂存的数据提交到数据库此时才会真正插入数据 session.commit()执行结果补充sqlalchemy2.0语法# sqlalchemy2.0语法 from sqlalchemy import insert # 构建insert语句 stmt insert(User).values([ {username: zhengshi, password: aaa}, {username: chenshi, password: bbb}, ]) # 执行语句 session.execute(stmt) session.commit()执行结果3、Delete删除数据删除数据的步骤查询数据 → 删除实例 → 提交会话删除操作会直接删除 MySQL 中的对应数据。from sqlalchemy_config import SessionFactory from base_model import User #导入Model模型 from sqlalchemy import delete with SessionFactory() as session: # 1. 单条数据删除 user session.query(User).filter(User.username zhangsan).first() if user: session.delete(user) # 对应 DELETE FROM user WHERE username wangwu session.commit() # 2. 批量删除对应 MySQL 的 DELETE ... WHERE ... # 删除所有 id 3 的用户 session.query(User).filter(User.id 3).delete() session.commit()执行结果补充sqlalchemy2.0语法# 3. sqlalchemy2.0语法 from sqlalchemy import delete # 构建 delete 语句 stmt delete(User).where(User.id 4) # 执行语句 session.execute(stmt) session.commit()4、Update更新数据更新数据的步骤查询数据 → 修改实例属性 → 提交会话更新操作会直接同步到 MySQL 数据库。from sqlalchemy_config import SessionFactory from base_model import User #导入Model模型 with SessionFactory() as session: # 1. 单条数据更新 user session.query(User).filter(User.username lisi).first() if user: user.password new_password123 # 修改密码对应 UPDATE user SET password new_password123 session.commit() # 提交更新 print(f{user})批量更新可以自行尝试查看结果# 2. 批量更新高效对应 MySQL 的 UPDATE ... WHERE ... # 将所有用户名以 w 开头的用户密码更新为 common_password session.query(User).filter(User.username.like(w%)).update({password: common_password}) session.commit()