一篇文章带你回顾完整SQL Alchemy

前言：

SQLAlchemy 是 Python 中最流行的数据库工具库，分为两大核心部分：

ORM（对象关系映射 Object Relational Mapper）【最常用】

Core（底层 SQL 表达式工具）

什么是ORM？

一种编程技术，用于实现面向对象编程语言中的对象与关系型数据库中的表之间的映射。

什么是Core？

他是SQL 语句构造器（表达式语言），是 SQLAlchemy 的底层基础，不依赖 ORM、不需要定义模型类；实现用代码动态组装标准 SQL，自动处理参数，防止 SQL 注入。

• 思想：直接构造表、字段、SQL 语句

• 只定义表结构 Table，没有对象映射

• 操作对象：表、列、查询表达式

• 返回：原始行数据（元组 / 字典）

SQL Alchemy重要组件

基础实现

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，则数据库准备完成。

TIPS！！！

sql代码需要通过 ；（分号） 结尾，不然判断不到语句是否结束

1、创建Engine（连接数据库）

这一步实现的主要功能就是连接到刚刚创建的数据库表

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine

# 1. 定义MySQL连接字符串（格式：数据库驱动://用户名:密码@主机:端口/数据库名?参数）
# 以pymysql驱动为例
db_url = "mysql+pymysql://root:123456@localhost:3306/sqlalchemy_demo?charset=utf8mb4"

# 2. 创建Engine（echo=True表示打印执行的SQL语句，方便调试，生产环境可关闭）
engine = create_engine(db_url, echo=True)

说明：

• root：数据库用户名，替换为自己的数据库账号；

• 123456：数据库密码，替换为自己的密码；

• sqlalchemy_demo：要连接的数据库名（需提前在MySQL中创建，或后续通过代码创建）；

• echo=True（可选）：开启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_key=True, autoincrement=True, comment="用户ID")
# 定义字段：username（用户名，非空，唯一）
username = Column(String(50), nullable=False, unique=True, comment="用户名")
# 定义字段：password（密码，非空）
password = Column(String(100), nullable=False, comment="密码")
# 定义字段：create_time（创建时间，默认当前时间）
create_time = Column(DateTime, default=datetime.now, comment="创建时间")

# 可选：定义__repr__方法，方便打印实例时查看信息
def __repr__(self):
return f"User(用户id={self.id}, 用户名={self.username}, 创建时间={self.create_time})"

其中：

Base为通用基类，后续添加的所有模型类都必须继承Base

模型类用于实现映射关系，必须与数据库表中字段对应

字段约束说明（常用）：

• primary_key=True：设为主键；

• autoincrement=True：自增（仅整数主键可用）；

• nullable=False：非空约束；

• unique=True：唯一约束；

• default：默认值；

• comment：字段注释。

定义__repr__方法的作用：

3、创建数据库表

通过Base类的create_all()方法，自动根据模型类创建数据库表（如果表已存在，则不会重复创建）：

# 基于Base类创建所有模型对应的数据库表（绑定Engine）
Base.metadata.create_all(engine)

print("数据库表创建成功！")

CURD增删改查

Session是与数据库交互的核心，所有操作都需通过Session完成，步骤为：创建Session → 执行操作 → 提交事务 → 关闭Session

0、抽取通用配置

为了更清楚的进行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 = "mysql+pymysql://root:123456@localhost:3306/sqlalchemy_demo?charset=utf8mb4"

# 2. 创建Engine（echo=True表示打印执行的SQL语句，方便调试，生产环境可关闭）
engine = create_engine(db_url, echo=True)
# 3. 创建Session工厂（绑定Engine）
SessionFactory = sessionmaker(bind=engine)
# 4. 创建Base基类（所有模型类必须继承此类）
Base = declarative_base()

模型类  —>   base_model

from 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_key=True, autoincrement=True, comment="用户ID")
# 定义字段：username（用户名，非空，唯一）
username = Column(String(50), nullable=False, unique=True, comment="用户名")
# 定义字段：password（密码，非空）
password = Column(String(100), nullable=False, comment="密码")
# 定义字段：create_time（创建时间，默认当前时间）
create_time = Column(DateTime, default=datetime.now, comment="创建时间")

# 可选：定义__repr__方法，方便打印实例时查看信息
def __repr__(self):
return f"User(用户id={self.id}, 用户名={self.username}, 创建时间={self.create_time})"

CRUD操作   —->    main

1、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 username="zhangsan" AND password="123456"')
user7 = session.query(User).filter_by(username='zhangsan', password='123456').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)查询所有列 聚合查询只能查询分组列与聚合函数
# 所以需要查询指定列
group1=session.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(username="lisi2", password="123456")
user2 = User(username="wangwu2", password="654321")
# 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()