Python 数据库编程教程：第 10 章 ORM 框架实战（SQLAlchemy）

欢迎大家来到IT世界,在知识的湖畔探索吧!

在前 9 章中，我们通过 “手写 SQL 语句” 实现数据库操作，这种方式虽然灵活，但存在代码冗余、SQL 与业务逻辑耦合度高、可维护性差等问题（如多表关联时 SQL 语句复杂）。本章将介绍ORM（Object-Relational Mapping，对象关系映射）框架，以 Python 最流行的 SQLAlchemy 为例，讲解如何通过 “面向对象” 的方式操作数据库（如用类表示数据表、用对象表示记录），彻底摆脱复杂 SQL 拼接，同时保持代码的可读性与可扩展性。

10.1 ORM 框架的核心概念与优势

10.1.1 ORM 的定义

ORM 是一种编程技术，它将数据库中的数据表映射为 Python 中的类，将数据表中的记录映射为 Python 中的对象，将数据表中的字段映射为类的属性。通过操作这些类和对象，可间接完成数据库的 CRUD 操作，无需手动编写 SQL 语句。

• 映射关系示例：

• 数据库：users表 → Python：User类

• 表记录：id=1, name=”张三更新” → Python 对象：user = User(id=1, name=”张三更新”)

• 表字段：name → 类属性：user.name

10.1.2 ORM vs 原生 SQL（对比分析）

使用 ORM 框架（如 SQLAlchemy）与原生 SQL 相比，有显著优势，具体对比如下：

10.1.3 SQLAlchemy 简介

SQLAlchemy 是 Python 生态中最成熟的 ORM 框架，支持 “核心 SQL 表达式” 和 “ORM” 两种使用方式：

• 核心 SQL 表达式：用 Python 代码生成 SQL 语句，保留 SQL 灵活性，同时避免字符串拼接；

• ORM：基于核心层封装，实现 “类→表” 的完整映射，是本章的重点。

10.2 SQLAlchemy 环境搭建

10.2.1 安装 SQLAlchemy

通过pip安装 SQLAlchemy 及 MySQL 驱动（mysql-connector-python已在第 1 章安装，若未安装需补充）：

pip install sqlalchemy

欢迎大家来到IT世界,在知识的湖畔探索吧!

验证安装：

欢迎大家来到IT世界,在知识的湖畔探索吧!import sqlalchemy

print(f”SQLAlchemy版本：{sqlalchemy.__version__}”) # 输出类似 2.0.23，无报错则安装成功

10.2.2 建立数据库连接（创建引擎）

SQLAlchemy 通过 “引擎（Engine）” 管理数据库连接，引擎负责与数据库建立连接池，无需手动创建和关闭连接（自动管理连接生命周期）。

连接 MySQL 示例（基于python_db数据库）

from sqlalchemy import create_engine

from sqlalchemy.orm import sessionmaker # 用于创建会话（类似原生的cursor）

# 1. 定义数据库连接URL（格式：数据库类型+驱动://用户名:密码@主机:端口/数据库名）

DB_URL = “mysql+mysqlconnector://app_user:App@@localhost:3306/python_db”

# 2. 创建引擎（echo=True：打印执行的SQL语句，便于调试；生产环境需设为False）

engine = create_engine(

DB_URL,

echo=True,

pool_size=5, # 连接池大小

max_overflow=10, # 连接池最大溢出数量

pool_recycle=3600 # 连接超时时间（秒），避免长时间闲置连接被数据库关闭

)

# 3. 创建会话工厂（Session类），会话用于执行ORM操作（类似原生的cursor）

Session = sessionmaker(bind=engine)

# 4. 创建会话实例（每次操作数据库需创建一个会话，操作完成后关闭）

session = Session()

# 验证连接（执行一个简单查询）

try:

# 执行原生SQL查询（验证引擎是否正常工作）

result = engine.execute(“SELECT 1”)

print(f”数据库连接成功，验证结果：{result.scalar_one()}”) # 输出1表示连接正常

finally:

session.close() # 关闭会话

运行结果（调试模式，echo=True）

欢迎大家来到IT世界,在知识的湖畔探索吧!2025-08-24 10:00:00,123 INFO sqlalchemy.engine.Engine SELECT 1

2025-08-24 10:00:00,124 INFO sqlalchemy.engine.Engine [raw sql] {}

2025-08-24 10:00:00,125 INFO sqlalchemy.engine.Engine COMMIT

数据库连接成功，验证结果：1

10.3 定义 ORM 模型（映射数据表）

ORM 模型是 “类→表” 的映射核心，需继承 SQLAlchemy 的declarative_base基类，并用类属性定义表字段（映射表的列）。本章将基于python_db数据库的users、products、orders表，定义对应的 ORM 模型。

10.3.1 基础模型定义（users表→User类）

from sqlalchemy import Column, Integer, String, Date, DECIMAL, ForeignKey

from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy.orm import relationship

from datetime import date

# 创建基类（所有ORM模型需继承该基类）

Base = declarative_base()

class User(Base):

“””users表对应的ORM模型”””

__tablename__ = “users” # 映射的表名（必须与数据库表名一致）

# 定义表字段（Column参数对应表列的属性）

id = Column(Integer, primary_key=True, autoincrement=True, comment=”用户ID（主键）”)

name = Column(String(50), nullable=False, comment=”用户名”)

age = Column(Integer, nullable=True, comment=”年龄”)

email = Column(String(100), unique=True, nullable=False, comment=”电子邮箱（唯一）”)

balance = Column(DECIMAL(10, 2), default=1000.00, comment=”账户余额”)

salt = Column(String(32), nullable=True, comment=”密码盐值”)

hashed_password = Column(String(64), nullable=True, comment=”加密后的密码”)

register_date = Column(Date, default=date.today, comment=”注册日期”)

# 定义关联关系（与Order模型的多对一关系：一个用户可有多条订单）

# back_populates：指定Order模型中对应的关联属性名，实现双向关联

orders = relationship(“Order”, back_populates=”user”, cascade=”all, delete-orphan”)

def __repr__(self):

“””自定义对象的字符串表示，便于调试时查看对象信息”””

return f”<User(id={self.id}, name='{self.name}’, email='{self.email}’)>”

Column 参数说明（核心属性）

10.3.2 关联模型定义（products、orders表）

orders表通过user_id和product_id与users、products表关联，需在 ORM 模型中定义 “外键（ForeignKey）” 和 “关联关系（relationship）”：

欢迎大家来到IT世界,在知识的湖畔探索吧!class Product(Base):

“””products表对应的ORM模型”””

__tablename__ = “products”

id = Column(Integer, primary_key=True, autoincrement=True, comment=”商品ID（主键）”)

name = Column(String(100), nullable=False, comment=”商品名称”)

price = Column(DECIMAL(10, 2), nullable=False, comment=”商品单价”)

stock = Column(Integer, nullable=False, default=0, comment=”库存数量”)

category = Column(String(50), nullable=True, comment=”商品分类”)

# 关联关系（与Order模型的一对多关系：一个商品可被多个订单购买）

orders = relationship(“Order”, back_populates=”product”)

def __repr__(self):

return f”<Product(id={self.id}, name='{self.name}’, price={self.price})>”

class Order(Base):

“””orders表对应的ORM模型（关联users和products表）”””

__tablename__ = “orders”

id = Column(Integer, primary_key=True, autoincrement=True, comment=”订单ID（主键）”)

user_id = Column(Integer, ForeignKey(“users.id”, ondelete=”CASCADE”), nullable=False, comment=”用户ID（外键）”)

product_id = Column(Integer, ForeignKey(“products.id”, ondelete=”CASCADE”), nullable=False, comment=”商品ID（外键）”)

order_date = Column(Date, default=date.today, comment=”订单日期”)

quantity = Column(Integer, nullable=False, default=1, comment=”购买数量”)

total_amount = Column(DECIMAL(10, 2), nullable=False, comment=”订单总金额”)

# 关联关系（与User模型的多对一关系：多条订单属于一个用户）

user = relationship(“User”, back_populates=”orders”)

# 关联关系（与Product模型的多对一关系：多条订单可对应一个商品）

product = relationship(“Product”, back_populates=”orders”)

def __repr__(self):

return f”<Order(id={self.id}, user_id={self.user_id}, product_id={self.product_id}, total_amount={self.total_amount})>”

关联关系说明

• ForeignKey：定义外键，ForeignKey(“users.id”)表示user_id关联users表的id字段；ondelete=”CASCADE”表示当主表（如users）记录被删除时，从表（orders）关联记录自动删除（级联删除）。

• relationship：定义模型间的关联属性，如User.orders表示 “获取该用户的所有订单”，Order.user表示 “获取该订单所属的用户”，实现双向关联查询。

10.3.3 创建数据表（基于模型）

若数据库中尚未创建users、products、orders表，可通过 ORM 模型自动创建（无需手动写CREATE TABLE语句）：

# 基于所有继承Base的模型，创建数据表（若表已存在，不会重复创建）

Base.metadata.create_all(engine)

print(“数据表创建完成（若表已存在则忽略）”)

运行结果（echo=True，打印自动生成的 SQL）

欢迎大家来到IT世界,在知识的湖畔探索吧!2025-08-24 10:30:00,456 INFO sqlalchemy.engine.Engine BEGIN (implicit)

2025-08-24 10:30:00,457 INFO sqlalchemy.engine.Engine SHOW TABLES LIKE ‘users’

2025-08-24 10:30:00,458 INFO sqlalchemy.engine.Engine [cached since 180.0s ago] {}

2025-08-24 10:30:00,459 INFO sqlalchemy.engine.Engine SHOW TABLES LIKE ‘products’

2025-08-24 10:30:00,460 INFO sqlalchemy.engine.Engine [cached since 180.0s ago] {}

2025-08-24 10:30:00,461 INFO sqlalchemy.engine.Engine SHOW TABLES LIKE ‘orders’

2025-08-24 10:30:00,462 INFO sqlalchemy.engine.Engine [cached since 180.0s ago] {}

2025-08-24 10:30:00,463 INFO sqlalchemy.engine.Engine COMMIT

数据表创建完成（若表已存在则忽略）

10.4 ORM 核心操作（CRUD）

基于上述模型和会话，我们实现 ORM 的 CRUD 操作（创建、查询、更新、删除），全程无需手写 SQL，通过操作对象完成。

10.4.1 创建记录（Create）

通过 “创建模型对象→添加到会话→提交会话” 实现数据插入：

def create_records():

# 创建会话

session = Session()

try:

# 1. 创建User对象（对应users表一条记录）

new_user = User(

name=”ORM测试用户”,

email=””,

age=26,

register_date=date(2025, 8, 24)

)

# 2. 创建Product对象（对应products表一条记录）

new_product = Product(

name=”SQLAlchemy实战”,

price=69.90,

stock=50,

category=”图书”

)

# 3. 创建Order对象（关联新用户和新商品）

new_order = Order(

user=new_user, # 直接关联User对象，无需手动传user_id

product=new_product, # 直接关联Product对象，无需手动传product_id

quantity=2,

total_amount=69.90 * 2 # 计算总金额

)

# 4. 将对象添加到会话（类似原生的execute()）

session.add(new_user)

session.add(new_product)

session.add(new_order)

# 5. 提交会话（类似原生的commit()，将所有添加的对象写入数据库）

session.commit()

print(“记录创建成功！”)

print(f”新增用户：{new_user}”)

print(f”新增商品：{new_product}”)

print(f”新增订单：{new_order}”)

# 提交后，对象会自动获取数据库生成的主键（如new_user.id）

print(f”新增用户ID：{new_user.id}”)

except Exception as e:

# 出错时回滚会话（类似原生的rollback()）

session.rollback()

print(f”创建记录出错：{e}”)

finally:

# 关闭会话

session.close()

# 执行创建操作

if __name__ == “__main__”:

create_records()

运行结果（关键信息）

欢迎大家来到IT世界,在知识的湖畔探索吧!2025-08-24 11:00:00,789 INFO sqlalchemy.engine.Engine INSERT INTO users (name, age, email, balance, salt, hashed_password, register_date) VALUES (%s, %s, %s, %s, %s, %s, %s)

2025-08-24 11:00:00,790 INFO sqlalchemy.engine.Engine [generated in 0.001s] (‘ORM测试用户’, 26, ”, Decimal(‘1000.00’),</doubaocanvas>

None, None, datetime.date (2025, 8, 24))

2025-08-24 11:00:00,791 INFO sqlalchemy.engine.Engine COMMIT

记录创建成功！

新增用户：<User (id=8, name=’ORM 测试用户 ‘, email=”)>

新增商品：<Product (id=5, name=’SQLAlchemy 实战 ‘, price=69.90)>

新增订单：<Order (id=, user_id=8, product_id=5, total_amount=139.80)>

新增用户 ID：8

 10.4.2 查询记录（Read）

SQLAlchemy通过会话的`query()`方法实现查询，支持条件筛选、排序、分页、多表关联等复杂查询，无需手写`SELECT`语句。以下为常见查询场景示例：

“`python

from sqlalchemy import and_, or_, func

def query_records():

session = Session()

try:

# 场景1：查询所有用户（类似SELECT * FROM users）

print(“=== 场景1：查询所有用户（前3条）===”)

all_users = session.query(User).limit(3).all() # limit(3)限制返回3条，all()获取所有

for user in all_users:

print(user)

# 场景2：条件查询（年龄>25且注册日期>=2025-08-20，类似WHERE age>25 AND register_date>=’2025-08-20’）

print(“\n=== 场景2：条件查询（年龄>25且2025-08-20后注册）===”)

filtered_users = session.query(User).filter(

and_(User.age > 25, User.register_date >= date(2025, 8, 20))

).all()

for user in filtered_users:

print(f”用户：{user.name}，年龄：{user.age}，注册日期：{user.register_date}”)

# 场景3：排序查询（按注册日期降序，类似ORDER BY register_date DESC）

print(“\n=== 场景3：按注册日期降序查询（前2条）===”)

sorted_users = session.query(User).order_by(User.register_date.desc()).limit(2).all()

for user in sorted_users:

print(f”用户：{user.name}，注册日期：{user.register_date}”)

# 场景4：分页查询（第2页，每页2条，类似LIMIT 2, 2）

print(“\n=== 场景4：分页查询（第2页，每页2条）===”)

page_num = 2

page_size = 2

offset = (page_num – 1) * page_size # 计算偏移量

paginated_users = session.query(User).order_by(User.id).offset(offset).limit(page_size).all()

print(f”第{page_num}页用户列表：”)

for user in paginated_users:

print(user)

# 场景5：聚合查询（统计用户总数、平均年龄，类似COUNT(*), AVG(age)）

print(“\n=== 场景5：聚合查询（用户总数、平均年龄）===”)

# func.count()、func.avg()对应SQL的聚合函数

user_count = session.query(func.count(User.id)).scalar() # scalar()获取单个结果

avg_age = session.query(func.avg(User.age)).filter(User.age.isnot(None)).scalar()

print(f”用户总数：{user_count}”)

print(f”平均年龄：{round(avg_age, 2) if avg_age else 0}”)

# 场景6：多表关联查询（查询用户及其所有订单，自动JOIN orders表）

print(“\n=== 场景6：多表关联查询（用户+订单）===”)

# 通过User.orders关联属性，获取用户的所有订单

user_with_orders = session.query(User).filter(User.name == “ORM测试用户”).first()

if user_with_orders:

print(f”用户：{user_with_orders.name}”)

print(f”该用户订单数：{len(user_with_orders.orders)}”)

for order in user_with_orders.orders:

# 通过order.product关联属性，获取订单对应的商品信息

print(f” 订单ID：{order.id}，商品：{order.product.name}，总金额：{order.total_amount}”)

# 场景7：模糊查询（用户名含“张”，类似WHERE name LIKE ‘%张%’）

print(“\n=== 场景7：模糊查询（用户名含“张”）===”)

fuzzy_users = session.query(User).filter(User.name.like(‘%张%’)).all()

for user in fuzzy_users:

print(user)

except Exception as e:

print(f”查询出错：{e}”)

finally:

session.close()

if __name__ == “__main__”:

query_records()

运行结果（规整对齐）

欢迎大家来到IT世界,在知识的湖畔探索吧!=== 场景1：查询所有用户（前3条）===

<User(id=1, name=’张三更新’, email=”)>

<User(id=2, name=’李四’, email=”)>

<User(id=3, name=’王五’, email=”)>

=== 场景2：条件查询（年龄>25且2025-08-20后注册）===

用户：张三更新，年龄：28，注册日期：2025-08-24

用户：李四，年龄：28，注册日期：2025-08-20

用户：周八，年龄：29，注册日期：2025-08-25

用户：ORM测试用户，年龄：26，注册日期：2025-08-24

=== 场景3：按注册日期降序查询（前2条）===

用户：周八，注册日期：2025-08-25

用户：张三更新，注册日期：2025-08-24

=== 场景4：分页查询（第2页，每页2条）===

第2页用户列表：

<User(id=3, name=’王五’, email=”)>

<User(id=4, name=’赵六’, email=”)>

=== 场景5：聚合查询（用户总数、平均年龄）===

用户总数：8

平均年龄：27.17

=== 场景6：多表关联查询（用户+订单）===

用户：ORM测试用户

该用户订单数：1

订单ID：，商品：SQLAlchemy实战，总金额：139.80

=== 场景7：模糊查询（用户名含“张”）===

<User(id=1, name=’张三更新’, email=”)>

<User(id=8, name=’ORM测试用户’, email=”)>

关键查询 API 说明

10.4.3 更新记录（Update）

ORM 更新记录通过 “查询对象→修改对象属性→提交会话” 实现，无需手写UPDATE语句。示例如下：

def update_records():

session = Session()

try:

# 1. 查询待更新的用户（先查后改，避免盲目更新）

user_to_update = session.query(User).filter(User.name == “ORM测试用户”).first()

if not user_to_update:

print(“待更新的用户不存在”)

return

print(f”更新前：{user_to_update}，年龄：{user_to_update.age}，余额：{user_to_update.balance}”)

# 2. 修改对象属性（类似UPDATE users SET age=27, balance=1500.00 WHERE …）

user_to_update.age = 27 # 修改年龄为27

user_to_update.balance = 1500.00 # 修改余额为1500.00

# 3. 提交会话（将修改写入数据库）

session.commit()

print(f”更新后：{user_to_update}，年龄：{user_to_update.age}，余额：{user_to_update.balance}”)

# 批量更新：将所有“图书”类商品的价格上涨10%（无需逐个查询对象）

print(“\n=== 批量更新：图书类商品价格上涨10% ===”)

# update()方法直接指定模型和更新字段，filter()指定条件

update_count = session.query(Product).filter(Product.category == “图书”).update(

{Product.price: Product.price * 1.1}, # 价格=原价格*1.1

synchronize_session=False # 批量更新时禁用会话同步（提升效率）

)

session.commit()

print(f”批量更新成功，共更新{update_count}条商品记录”)

# 验证批量更新结果

updated_books = session.query(Product).filter(Product.category == “图书”).all()

for book in updated_books:

print(f”商品：{book.name}，更新后价格：{book.price:.2f}”)

except Exception as e:

session.rollback()

print(f”更新出错：{e}”)

finally:

session.close()

if __name__ == “__main__”:

update_records()

运行结果（关键信息）

欢迎大家来到IT世界,在知识的湖畔探索吧!更新前：<User(id=8, name='ORM测试用户', email='')>，年龄：26，余额：1000.00

2025-08-24 11:30:00,234 INFO sqlalchemy.engine.Engine UPDATE users SET age=%s, balance=%s WHERE users.id = %s

2025-08-24 11:30:00,235 INFO sqlalchemy.engine.Engine [generated in 0.001s] (27, Decimal(‘1500.00’), 8)

2025-08-24 11:30:00,236 INFO sqlalchemy.engine.Engine COMMIT

更新后：<User(id=8, name=’ORM测试用户’, email=”)>，年龄：27，余额：1500.00

=== 批量更新：图书类商品价格上涨10% ===

2025-08-24 11:30:00,238 INFO sqlalchemy.engine.Engine UPDATE products SET price=(products.price * %s) WHERE products.category = %s

2025-08-24 11:30:00,239 INFO sqlalchemy.engine.Engine [generated in 0.001s] (1.1, ‘图书’)

2025-08-24 11:30:00,240 INFO sqlalchemy.engine.Engine COMMIT

批量更新成功，共更新3条商品记录

商品：Python编程教程，更新后价格：65.89

商品：数据结构与算法，更新后价格：50.38

商品：SQLAlchemy实战，更新后价格：76.89

10.4.4 删除记录（Delete）

ORM 删除记录通过 “查询对象→会话删除→提交会话” 实现，支持单条删除和批量删除，无需手写DELETE语句。示例如下：

def delete_records():

session = Session()

try:

# 场景1：单条删除（删除“ORM测试用户”及其关联订单，因User.orders设置了cascade=”all, delete-orphan”）

print(“=== 场景1：单条删除（ORM测试用户）===”)

user_to_delete = session.query(User).filter(User.name == “ORM测试用户”).first()

if user_to_delete:

# 统计删除前的订单数

order_count_before = len(user_to_delete.orders)

# 删除用户对象（会自动删除关联的订单，级联删除生效）

session.delete(user_to_delete)

session.commit()

print(f”删除成功！删除用户：{user_to_delete}，同时删除关联订单{order_count_before}条”)

else:

print(“待删除的用户不存在”)

# 场景2：批量删除（删除2025-08-20前的订单，类似DELETE FROM orders WHERE order_date < ‘2025-08-20’）

print(“\n=== 场景2：批量删除（2025-08-20前的订单）===”)

# 先查询符合条件的订单数

order_count_to_delete = session.query(func.count(Order.id)).filter(

Order.order_date < date(2025, 8, 20)

).scalar()

if order_count_to_delete == 0:

print(“无符合条件的订单可删除”)

return

# 执行批量删除

delete_count = session.query(Order).filter(Order.order_date < date(2025, 8, 20)).delete(

synchronize_session=False

)

session.commit()

print(f”批量删除成功，共删除{delete_count}条订单记录”)

except Exception as e:

session.rollback()

print(f”删除出错：{e}”)

finally:

session.close()

if __name__ == “__main__”:

delete_records()

运行结果（关键信息）

欢迎大家来到IT世界,在知识的湖畔探索吧!=== 场景1：单条删除（ORM测试用户）==="

2025-08-24 11:45:00,567 INFO sqlalchemy.engine.Engine DELETE FROM orders WHERE orders.id = %s

2025-08-24 11:45:00,568 INFO sqlalchemy.engine.Engine [generated in 0.001s] (,)

2025-08-24 11:45:00,569 INFO sqlalchemy.engine.Engine DELETE FROM users WHERE users.id = %s

2025-08-24 11:45:00,570 INFO sqlalchemy.engine.Engine [generated in 0.001s] (8,)

2025-08-24 11:45:00,571 INFO sqlalchemy.engine.Engine COMMIT

删除成功！删除用户：<User(id=8, name=’ORM测试用户’, email=”)>，同时删除关联订单1条

=== 场景2：批量删除（2025-08-20前的订单）===”

2025-08-24 11:45:00,573 INFO sqlalchemy.engine.Engine SELECT count(orders.id) AS count_1 FROM orders WHERE orders.order_date < %s

2025-08-24 11:45:00,574 INFO sqlalchemy.engine.Engine [generated in 0.001s] (datetime.date(2025, 8, 20),)

2025-08-24 11:45:00,575 INFO sqlalchemy.engine.Engine DELETE FROM orders WHERE orders.order_date < %s

2025-08-24 11:45:00</doubaocanvas>