SQLAlchemy 快速入门：Python 数据库操作实战 – wiki基地

作者： /

SQLAlchemy 快速入门：Python 数据库操作实战

在现代软件开发中，数据持久化是一个核心需求。无论是网站的用户数据、移动应用的配置信息，还是企业系统的业务记录，都需要存储在数据库中。Python 提供了多种与数据库交互的方式，其中最强大和灵活的库之一便是 SQLAlchemy。

SQLAlchemy 是一个为 Python SQL 数据库提供的工具包和对象关系映射（ORM）器。它提供了全功能的持久层，为开发者提供了极大的灵活性和控制力。与一些简单的数据库连接库不同，SQLAlchemy 不仅是一个简单的包装器，它抽象了数据库的大部分细节，让你可以用更“Pythonic”的方式来操作数据库，无论是编写原生的 SQL 语句（通过 Core），还是将 Python 对象映射到数据库表（通过 ORM）。

本篇文章旨在提供一个 SQLAlchemy 的快速入门指南，通过实际的代码示例，带你了解 SQLAlchemy 的核心概念、基本操作，以及如何使用它进行日常的数据库增删改查（CRUD）操作。我们将主要关注 SQLAlchemy 的两个主要组件：SQL Expression Language (Core) 和 ORM。

1. 为什么选择 SQLAlchemy？

在深入学习之前，让我们快速了解一下使用 SQLAlchemy 的优势：

  • 抽象化数据库差异： SQLAlchemy 支持多种数据库后端（如 PostgreSQL, MySQL, SQLite, Oracle, SQL Server 等），它为你处理不同数据库之间的 SQL 语法差异。你只需要使用统一的 SQLAlchemy 语法，就可以轻松切换数据库。
  • 强大的 SQL Expression Language (Core)： 如果你需要编写复杂的 SQL 语句，或者追求极致的性能，SQLAlchemy Core 允许你使用 Python 代码构建 SQL 表达式，这种方式比手动拼接字符串更安全（防止 SQL 注入）且更具可读性。
  • 灵活的 ORM： SQLAlchemy 的 ORM 允许你将数据库表映射到 Python 类。你可以像操作 Python 对象一样操作数据库记录，极大地提高了开发效率，减少了重复的 SQL 编写工作。它的映射方式非常灵活，支持复杂的关系模型。
  • 高性能： SQLAlchemy 在底层进行了大量优化，例如连接池、预编译语句等，以提高数据库操作的性能。
  • 成熟稳定： SQLAlchemy 是一个经过长时间发展和广泛应用的成熟库，拥有庞大的社区支持和完善的文档。

2. 环境准备与安装

开始之前，你需要确保安装了 Python。然后，通过 pip 安装 SQLAlchemy：

bash
pip install sqlalchemy

SQLAlchemy 本身是一个数据库抽象层，它需要对应的数据库驱动才能连接到特定的数据库。对于本入门教程，我们将使用 Python 内置的 SQLite 数据库，它无需额外安装驱动。如果你需要连接其他数据库，你需要安装相应的驱动：

  • PostgreSQL: pip install psycopg2-binary
  • MySQL: pip install mysql-connector-python 或 pip install PyMySQL
  • SQL Server: pip install pyodbc
  • Oracle: pip install cx_oracle (或 python-oracledb for newer versions)

3. SQLAlchemy Core 快速入门：直接操作 SQL 表达式

SQLAlchemy Core 提供了构建和执行 SQL 语句的 Pythonic 方式。它不涉及对象映射，更接近于直接编写 SQL，但通过 Python 结构来表示 SQL 元素，提供了更好的安全性和灵活性。

3.1 建立数据库连接 (Engine)

一切的起点是 Engine。Engine 是 SQLAlchemy 连接数据库的接口，它管理着连接池和其他与数据库通信相关的资源。

“`python
from sqlalchemy import create_engine

使用 SQLite 内存数据库，数据不会持久化

engine = create_engine(“sqlite:///:memory:”)

使用 SQLite 文件数据库，数据将保存在 mydatabase.db 文件中

如果文件不存在，SQLAlchemy 会自动创建

engine = create_engine(“sqlite:///mydatabase.db”)

print(“数据库引擎创建成功。”)
“`

数据库 URL 的格式通常是 dialect+driver://user:password@host:port/database。对于 SQLite 文件数据库，格式简化为 sqlite:///path/to/database.db。

3.2 定义表结构 (MetaData, Table, Column)

在 SQLAlchemy Core 中，我们使用 MetaData 对象来收集表结构信息，Table 对象表示数据库中的一个表，Column 对象表示表中的列。

“`python
from sqlalchemy import MetaData, Table, Column, Integer, String, ForeignKey

MetaData 是一个容器，用于收集数据库表的定义

metadata = MetaData()

定义一个 ‘users’ 表

users_table = Table(
“users”, # 表名
metadata, # 关联的 MetaData 对象
Column(“id”, Integer, primary_key=True), # id 列，整型，主键
Column(“name”, String(50), nullable=False), # name 列，最大长度50的字符串，不允许为空
Column(“fullname”, String), # fullname 列，字符串
# UniqueConstraint(‘name’, ‘fullname’, name=’uniq_name_fullname’) # 复合唯一约束示例
)

定义一个 ‘addresses’ 表，与 ‘users’ 表有关联 (外键)

addresses_table = Table(
“addresses”,
metadata,
Column(“id”, Integer, primary_key=True),
Column(“user_id”, Integer, ForeignKey(“users.id”), nullable=False), # 外键，关联 users 表的 id 列
Column(“email_address”, String(100), nullable=False),
# UniqueConstraint(‘user_id’, ’email_address’, name=’uniq_user_email’) # 复合唯一约束示例
)

print(“表结构定义完成。”)
“`

MetaData 对象会记录所有通过它定义的 Table 对象。

3.3 创建表

定义好表结构后，我们可以使用 metadata.create_all(engine) 来在数据库中创建这些表。

“`python

在数据库中创建所有在 metadata 中定义的表

checkfirst=True 表示创建前先检查表是否存在，避免重复创建报错

metadata.create_all(engine, checkfirst=True)

print(“数据库表创建成功 (如果不存在)。”)
“`

3.4 插入数据 (Insert)

使用 Core 插入数据，我们构建一个 insert() 表达式，并使用连接对象 Connection 来执行它。

“`python
from sqlalchemy import insert

获取数据库连接

推荐使用 ‘with’ 语句管理连接，确保连接在使用后被关闭

with engine.connect() as connection:
# 插入单条数据
insert_stmt = insert(users_table).values(name=”spongebob”, fullname=”Spongebob Squarepants”)
result = connection.execute(insert_stmt)
print(f”插入单条数据成功，新的用户ID是: {result.lastrowid}”) # lastrowid 获取自增主键的值

# 插入多条数据
insert_many_stmt = insert(users_table)
users_to_add = [
    {"name": "patrick", "fullname": "Patrick Star"},
    {"name": "sandy", "fullname": "Sandy Cheeks"},
]
result_many = connection.execute(insert_many_stmt, users_to_add)
print(f"插入多条数据成功，插入行数：{result_many.rowcount}")

# 插入地址数据
insert_addresses_stmt = insert(addresses_table).values([
    {"user_id": 1, "email_address": "[email protected]"}, # 假设 Spongebob 的 id 是 1
    {"user_id": 2, "email_address": "[email protected]"}, # 假设 Patrick 的 id 是 2
    {"user_id": 2, "email_address": "[email protected]"}, # Patrick 的另一个地址
    {"user_id": 3, "email_address": "[email protected]"}, # 假设 Sandy 的 id 是 3
])
connection.execute(insert_addresses_stmt)
print("插入地址数据成功。")

# 提交事务（在自动提交模式下可能不需要显式调用，但为了保险和清晰，手动提交是个好习惯）
# SQLite 默认是自动提交 DML (INSERT, UPDATE, DELETE)
# 对于其他数据库或需要多条语句在一个事务中执行时，connection.commit() 很重要
# connection.commit() # 在 with 块结束时，连接会自动关闭并提交或回滚未决事务

print(“所有插入操作完成。”)
“`

3.5 查询数据 (Select)

使用 Core 查询数据，我们构建一个 select() 表达式。

“`python
from sqlalchemy import select, func

with engine.connect() as connection:
# 查询所有用户
select_all_stmt = select(users_table)
result_all = connection.execute(select_all_stmt)
print(“\n查询所有用户:”)
for row in result_all:
print(row) # row 是一个 RowProxy 对象，可以通过索引或列名访问值

# 查询特定用户 (WHERE)
select_sandy_stmt = select(users_table).where(users_table.c.name == "sandy") # 使用 .c 访问列
result_sandy = connection.execute(select_sandy_stmt)
print("\n查询名字为 'sandy' 的用户:")
print(result_sandy.fetchone()) # fetchone() 获取一条结果，然后结果集被耗尽

# 查询部分列
select_names_stmt = select(users_table.c.name, users_table.c.fullname)
result_names = connection.execute(select_names_stmt)
print("\n查询所有用户的 name 和 fullname:")
for name, fullname in result_names: # 可以直接解包
    print(f"Name: {name}, Fullname: {fullname}")

# 条件查询 (AND, OR)
from sqlalchemy import and_, or_
select_complex_where_stmt = select(users_table).where(
    and_(users_table.c.name.startswith("s"), users_table.c.id > 0) # 使用 and_() 组合条件
).order_by(users_table.c.id) # 排序

print("\n查询名字以 's' 开头且 ID > 0 的用户 (按 ID 排序):")
for row in connection.execute(select_complex_where_stmt):
     print(row)

# 限制和偏移 (LIMIT, OFFSET)
select_limit_offset_stmt = select(users_table).limit(2).offset(1) # 跳过第一条，取两条
print("\n查询第二和第三个用户:")
for row in connection.execute(select_limit_offset_stmt):
    print(row)

# 聚合函数和分组 (COUNT, GROUP BY)
# 统计用户数量
count_stmt = select(func.count()).select_from(users_table)
user_count = connection.execute(count_stmt).scalar() # scalar() 获取查询结果的第一个列的第一个值
print(f"\n总用户数量: {user_count}")

# 统计每个用户的地址数量 (JOIN and GROUP BY)
# join() 方法默认使用外键关系进行连接
join_stmt = select(users_table.c.name, func.count(addresses_table.c.id).label("address_count")).\
    join(addresses_table).\
    group_by(users_table.c.name)

print("\n每个用户的地址数量:")
for name, count in connection.execute(join_stmt):
    print(f"用户: {name}, 地址数量: {count}")

print(“所有查询操作完成。”)
“`

3.6 更新数据 (Update)

使用 Core 更新数据，我们构建一个 update() 表达式。

“`python
from sqlalchemy import update

with engine.connect() as connection:
# 更新特定用户的全名
update_stmt = update(users_table).where(users_table.c.name == “spongebob”).values(fullname=”Spongebob H. Squarepants”)
result = connection.execute(update_stmt)
print(f”\n更新用户 ‘spongebob’ 的全名，影响行数: {result.rowcount}”)

# 更新所有用户的全名（慎用！）
# update_all_stmt = update(users_table).values(fullname="New Fullname")
# result_all = connection.execute(update_all_stmt)
# print(f"更新所有用户，影响行数: {result_all.rowcount}")

# 再次查询 'spongebob' 验证更新
select_spongebob_stmt = select(users_table).where(users_table.c.name == "spongebob")
print("\n更新后查询 'spongebob':")
print(connection.execute(select_spongebob_stmt).fetchone())

print(“所有更新操作完成。”)
“`

3.7 删除数据 (Delete)

使用 Core 删除数据，我们构建一个 delete() 表达式。

“`python
from sqlalchemy import delete

with engine.connect() as connection:
# 删除特定用户 (及其关联的地址，如果设置了级联删除的外键)
# 注意：SQLite 默认外键约束是禁用的，需要在连接URL中添加 ?foreign_keys=on 来启用
# 或者执行 PRAGMA foreign_keys = ON;
# 为了简单，这里我们先删除地址，再删除用户
delete_address_stmt = delete(addresses_table).where(addresses_table.c.user_id == 1) # 删除 Spongebob 的地址
result_address = connection.execute(delete_address_stmt)
print(f”\n删除用户 ID 1 的地址，影响行数: {result_address.rowcount}”)

delete_user_stmt = delete(users_table).where(users_table.c.name == "spongebob") # 删除 Spongebob 用户
result_user = connection.execute(delete_user_stmt)
print(f"删除用户 'spongebob'，影响行数: {result_user.rowcount}")

# 删除所有用户 (慎用！)
# delete_all_stmt = delete(users_table)
# result_all = connection.execute(delete_all_stmt)
# print(f"删除所有用户，影响行数: {result_all.rowcount}")

# 再次查询 'spongebob' 验证删除
select_spongebob_stmt = select(users_table).where(users_table.c.name == "spongebob")
print("\n删除后查询 'spongebob':")
print(connection.execute(select_spongebob_stmt).fetchone()) # 应该返回 None

print(“所有删除操作完成。”)
“`

SQLAlchemy Core 提供了一种强大且灵活的方式来构建和执行 SQL 语句，让你能够充分利用数据库的功能，同时享受 Python 带来的便利和安全性。然而，对于复杂的应用，特别是那些需要处理大量对象之间关系的应用，SQLAlchemy 的 ORM 部分通常是更受欢迎的选择。

4. SQLAlchemy ORM 快速入门：对象关系映射

SQLAlchemy ORM 允许你将 Python 类映射到数据库表，将类的实例映射到表中的行，将类的属性映射到列。这样，你可以像操作普通 Python 对象一样操作数据库数据。

4.1 定义映射类 (Declarative Base)

使用 ORM，我们通常采用声明式（Declarative）映射方式。首先，需要创建一个基类，然后定义继承自这个基类的 Python 类，这些类将自动映射到数据库表。

“`python
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, create_engine
from sqlalchemy.orm import relationship, sessionmaker

创建一个声明式基类

Base = declarative_base()

定义映射类 – Users 表

class User(Base):
tablename = “users_orm” # 数据库表名，与 Core 示例区分开

id = Column(Integer, primary_key=True)
name = Column(String(50), nullable=False)
fullname = Column(String)

# 定义与 Address 的一对多关系
# relationship() 表示这个 User 对象有一个名为 'addresses' 的属性
# 当访问 user.addresses 时，SQLAlchemy 会自动查询 Addresses 表中 user_id 等于 user.id 的记录
# back_populates 指定在 Address 类中对应 User 对象的属性名
addresses = relationship("Address", back_populates="user")

# __repr__ 方法方便打印对象时查看其内容
def __repr__(self):
    return f"<User(id={self.id}, name='{self.name}', fullname='{self.fullname}')>"

定义映射类 – Addresses 表

class Address(Base):
tablename = “addresses_orm” # 数据库表名

id = Column(Integer, primary_key=True)
email_address = Column(String(100), nullable=False)
user_id = Column(Integer, ForeignKey("users_orm.id"), nullable=False) # 外键关联 User 表

# 定义与 User 的多对一关系
# relationship() 表示这个 Address 对象有一个名为 'user' 的属性
# 当访问 address.user 时，SQLAlchemy 会自动查询 Users 表中 id 等于 address.user_id 的记录
# back_populates 指定在 User 类中对应 Address 对象的属性名
user = relationship("User", back_populates="addresses")

def __repr__(self):
    return f"<Address(id={self.id}, email_address='{self.email_address}', user_id={self.user_id})>"

print(“ORM 映射类定义完成。”)
“`

4.2 创建表 (使用 ORM MetaData)

声明式基类 Base 包含一个 metadata 属性，它实际上是一个 MetaData 对象，包含了所有通过继承 Base 定义的表结构。我们可以使用它来创建表。

“`python

使用与 Core 示例不同的数据库文件，以免冲突

engine_orm = create_engine(“sqlite:///myormdatabase.db”)

在数据库中创建所有在 Base.metadata 中定义的表

Base.metadata.create_all(engine_orm, checkfirst=True)

print(“ORM 数据库表创建成功 (如果不存在)。”)
“`

4.3 创建 Session

在 ORM 中，Session 是你与数据库交互的主要接口。Session 管理着你的对象的状态（例如，哪些对象是新的、哪些已被修改、哪些已被删除）以及与数据库的同步（提交、回滚）。它实现了单元工作模式（Unit of Work）。

通常，你会先创建一个 sessionmaker 工厂，然后用它创建具体的 Session 实例。

“`python

创建一个 Session 工厂

bind 参数指定了这个 Session 连接哪个数据库引擎

Session = sessionmaker(bind=engine_orm)

创建一个 Session 实例

推荐使用 ‘with’ 语句管理 Session 的生命周期，确保它在使用后被关闭，

并且在发生异常时能正确回滚。

session = Session()

print(“Session 工厂创建完成。”)
“`

4.4 插入数据 (Add)

使用 ORM 插入数据，我们创建映射类的实例，然后将实例添加到 Session 中，最后提交 Session。

“`python

使用 try…except…finally 结构来演示事务管理和 Session 关闭

或者使用 with Session() as session: (如果你的 SQLAlchemy 版本支持，或自定义context manager)

为了简化，这里先使用 try…finally 结构

session = Session()
try:
# 创建 User 对象
user1 = User(name=”spongebob”, fullname=”Spongebob Squarepants”)
user2 = User(name=”patrick”, fullname=”Patrick Star”)
user3 = User(name=”sandy”, fullname=”Sandy Cheeks”)

# 将对象添加到 Session 中
session.add(user1)
session.add_all([user2, user3]) # 添加多个对象

# 创建 Address 对象，并关联到 User 对象
# 可以通过直接赋值关系属性来建立关系
address1 = Address(email_address="[email protected]", user=user1)
address2 = Address(email_address="[email protected]", user=user2)
address3 = Address(email_address="[email protected]", user=user2) # Patrick 的另一个地址
address4 = Address(email_address="[email protected]", user=user3)

session.add_all([address1, address2, address3, address4])

# 或者，可以通过 User 对象的 addresses 属性来添加地址 (如果 back_populates 设置正确)
# user1.addresses.append(Address(email_address="[email protected]"))
# session.add(user1) # 如果 user1 已经在 session 中，这一步可以省略，Session 会跟踪其变化
# session.add(user1.addresses[-1]) # 需要将新创建的 Address 对象也添加到 session 中

# 提交事务，将 Session 中的变化同步到数据库
session.commit()
print("使用 ORM 插入数据成功。")

except Exception as e:
# 如果发生异常，回滚事务
session.rollback()
print(f”插入数据失败，已回滚事务: {e}”)
finally:
# 关闭 Session
session.close()
print(“Session 已关闭。”)

“`

4.5 查询数据 (Query)

使用 ORM 查询数据，我们使用 session.query() 方法，传递需要查询的映射类。

“`python
session = Session()
try:
# 查询所有用户
print(“\n查询所有用户:”)
users = session.query(User).all() # .all() 执行查询并返回所有结果的列表
for user in users:
print(user)
# 也可以访问关联的地址 (会触发额外的数据库查询，称为懒加载 lazy loading)
# print(f” 地址: {[addr.email_address for addr in user.addresses]}”)

# 查询特定用户 (filter_by)
print("\n查询名字为 'sandy' 的用户:")
sandy_user = session.query(User).filter_by(name="sandy").first() # filter_by 用于简单的等于条件，.first() 获取第一条结果或 None
print(sandy_user)
# 访问关联的地址
if sandy_user:
     print(f"  地址: {[addr.email_address for addr in sandy_user.addresses]}")


# 条件查询 (filter)
from sqlalchemy import and_ # 也可以直接使用 & 运算符
print("\n查询名字以 'p' 开头且 ID > 0 的用户 (按 ID 排序):")
# filter() 用于更复杂的条件，可以使用类属性和比较运算符
filtered_users = session.query(User).\
    filter(and_(User.name.startswith("p"), User.id > 0)).\
    order_by(User.id).\
    all()
for user in filtered_users:
    print(user)

# 限制和偏移 (limit, offset)
print("\n查询第二和第三个用户:")
limited_users = session.query(User).limit(2).offset(1).all()
for user in limited_users:
    print(user)

# 查询部分列
print("\n查询所有用户的 name 和 fullname:")
name_fullname_results = session.query(User.name, User.fullname).all()
for name, fullname in name_fullname_results:
     print(f"Name: {name}, Fullname: {fullname}")

# 聚合函数和分组
print(f"\n总用户数量: {session.query(func.count(User.id)).scalar()}") # func.count() 使用 ORM

# 使用 join 查询 (关联查询)
print("\n查询拥有 '[email protected]' 地址的用户:")
# 可以直接使用类进行 join
user_with_specific_address = session.query(User).\
    join(Address).\
    filter(Address.email_address == '[email protected]').\
    first()
print(user_with_specific_address)

# 查询每个用户的地址数量
print("\n每个用户的地址数量:")
address_counts = session.query(User.name, func.count(Address.id).label("address_count")).\
    join(Address).\
    group_by(User.name).\
    all()
for name, count in address_counts:
    print(f"用户: {name}, 地址数量: {count}")

except Exception as e:
print(f”查询数据失败: {e}”)
finally:
session.close()
print(“Session 已关闭。”)
“`

ORM 的查询语法更加面向对象，直接使用类和属性进行过滤、排序等操作。

4.6 更新数据 (Modify and Commit)

使用 ORM 更新数据，我们首先通过查询获取到要更新的对象，然后直接修改对象的属性，最后提交 Session。Session 会自动跟踪对象的修改并生成相应的 UPDATE 语句。

“`python
session = Session()
try:
# 查询要更新的对象
spongebob_user = session.query(User).filter_by(name=”spongebob”).first()

if spongebob_user:
    # 修改对象的属性
    spongebob_user.fullname = "Spongebob H. Squarepants (Updated)"
    print(f"\n修改用户 'spongebob' 的全名到: {spongebob_user.fullname}")

    # 提交 Session，将修改同步到数据库
    session.commit()
    print("使用 ORM 更新数据成功。")
else:
    print("\n未找到用户 'spongebob' 进行更新。")

except Exception as e:
session.rollback() # 如果更新失败，回滚事务
print(f”更新数据失败，已回滚事务: {e}”)
finally:
session.close()
print(“Session 已关闭。”)

再次查询验证更新

session = Session()
try:
print(“\n更新后查询 ‘spongebob’:”)
updated_spongebob = session.query(User).filter_by(name=”spongebob”).first()
print(updated_spongebob)
except Exception as e:
print(f”查询验证失败: {e}”)
finally:
session.close()
“`

4.7 删除数据 (Delete)

使用 ORM 删除数据，我们通过查询获取要删除的对象，然后将对象添加到 Session 的删除列表中，最后提交 Session。

“`python
session = Session()
try:
# 查询要删除的对象
spongebob_user = session.query(User).filter_by(name=”spongebob”).first()

if spongebob_user:
    # 将对象标记为待删除
    session.delete(spongebob_user)
    print(f"\n标记用户 {spongebob_user} 为待删除。")

    # 提交 Session，执行删除操作
    session.commit()
    print("使用 ORM 删除数据成功。")

    # 再次查询 'spongebob'，应该返回 None
    print("\n删除后再次查询 'spongebob':")
    deleted_user = session.query(User).filter_by(name="spongebob").first()
    print(f"查询结果: {deleted_user}")

else:
    print("\n未找到用户 'spongebob' 进行删除。")

except Exception as e:
session.rollback() # 如果删除失败，回滚事务
print(f”删除数据失败，已回滚事务: {e}”)
finally:
session.close()
print(“Session 已关闭。”)
“`

请注意，如果你的数据库外键设置了级联删除（ON DELETE CASCADE），删除 User 对象时会自动删除关联的 Address 对象。如果未设置，你可能需要手动删除关联的 Address 对象或者处理数据库抛出的外键约束错误。在 ORM 层面，SQLAlchemy 的 relationship 参数也可以配置级联删除行为（如 cascade="all, delete-orphan"）。

5. Core 与 ORM 的选择

SQLAlchemy Core 和 ORM 都有其适用场景：

  • 选择 Core 的情况：
    • 你需要编写高度优化的、复杂的 SQL 语句，而 ORM 的抽象可能成为限制。
    • 你正在处理一个已经存在的、结构复杂的数据库，直接使用 Core 可能比建立完整的 ORM 映射更容易。
    • 你只需要执行简单的、一次性的查询或操作。
    • 你希望对底层 SQL 有更直接的控制。
  • 选择 ORM 的情况：
    • 你的应用是面向对象的，希望将数据库操作集成到对象模型中。
    • 你需要频繁地处理对象之间的复杂关系（一对多、多对多）。
    • 你希望减少手动编写 SQL 的工作量，提高开发效率。
    • 你希望利用 SQLAlchemy ORM 提供的 Session 管理、Identity Map 等高级特性。

在实际应用中，很多项目会同时使用 Core 和 ORM。你可以用 ORM 处理大部分日常的 CRUD 操作和关系管理，而用 Core 来处理少数需要精细控制或性能优化的复杂查询。

6. 事务管理与 Session 生命周期

无论是 Core 的 Connection 还是 ORM 的 Session，它们都涉及到事务管理。

  • Core 的 Connection： 当你通过 engine.connect() 获取一个连接时，通常会自动开始一个事务。执行 DML (INSERT, UPDATE, DELETE) 语句后，需要显式调用 connection.commit() 来提交事务，将更改永久保存到数据库。如果发生错误，调用 connection.rollback() 来撤销事务中的所有更改。使用 with engine.connect() as connection: 语句可以更好地管理连接的生命周期，with 块结束时，连接会自动关闭，并根据执行情况自动提交或回滚未决事务（取决于配置和操作类型）。
  • ORM 的 Session： Session 管理着一个或多个事务。当你调用 session.add()、session.delete() 或修改对象的属性时，这些更改首先发生在 Session 的内部状态中（Unit of Work），并不会立即同步到数据库。只有当你调用 session.commit() 时，Session 才会将这些累积的更改批量地刷新（flush）到数据库，并提交底层的事务。如果发生错误，调用 session.rollback() 来撤销 Session 中的所有更改（包括那些已经刷新但未提交到数据库的），并将对象状态恢复到上一次提交后的状态。使用 session.close() 来释放 Session 占用的资源，特别是数据库连接。同样，推荐使用 with session_scope() as session: 这样的模式来管理 Session 的生命周期，确保在任何情况下都能正确地提交、回滚或关闭 Session。虽然 SQLAlchemy 本身没有内置一个简单的 with Session(...) 块，但可以通过编写一个简单的上下文管理器实现，或者使用一些第三方库（如 sqlalchemy.orm.session.close_all_sessions 或社区推荐的模式）。对于入门，理解 commit(), rollback(), close() 的作用是关键。

“`python

简单的 Session 上下管理器示例 (实际项目中可能会用更完善的版本)

from contextlib import contextmanager

@contextmanager
def session_scope(Session):
“””Provide a transactional scope around a series of operations.”””
session = Session()
try:
yield session
session.commit()
except:
session.rollback()
raise
finally:
session.close()

如何使用上面定义的上下文管理器

with session_scope(Session) as session:

# 在这里执行 ORM 操作

new_user = User(name=”test_user”)

session.add(new_user)

# with 块结束时，如果无异常则提交，有异常则回滚并重新抛出，最后关闭 session

``
在快速入门阶段，理解session.add(),session.commit(),session.rollback(),session.close()` 是基础。更高级的应用会深入探讨 Session 的各种状态和生命周期管理。

7. 总结与下一步

恭喜你！通过本文的学习，你已经初步掌握了 SQLAlchemy 的核心概念和基本用法，包括：

  • 创建数据库引擎 (create_engine)
  • 使用 Core 定义表结构 (MetaData, Table, Column) 和进行 CRUD 操作 (insert, select, update, delete)
  • 使用 ORM 定义映射类 (declarative_base, 类定义) 和进行 CRUD 操作 (Session, add, query, commit, rollback, delete)
  • 了解了 Core 和 ORM 的适用场景以及基本的事务管理概念。

这仅仅是 SQLAlchemy 强大功能的冰山一角。要成为熟练的 SQLAlchemy 用户，你还需要进一步学习：

  • 更复杂的查询技术：子查询、联合查询、CTE (Common Table Expressions) 等。
  • 更多的关系类型：多对多关系。
  • 级联操作（Cascades）：例如删除主对象时自动删除关联对象。
  • 加载策略（Loading Strategies）：如何优化关联对象的加载（懒加载、预加载、即时加载）。
  • 事件系统（Events）：在特定操作发生时触发自定义代码。
  • 连接池配置：如何优化数据库连接的使用。
  • 数据库迁移工具：如 Alembic，用于管理数据库模式的变化。

强烈推荐查阅 SQLAlchemy 的官方文档，它是最权威和详细的学习资源。通过不断实践和深入学习，你将能够充分发挥 SQLAlchemy 的潜力，构建健壮、高效的数据库应用。

希望这篇详细的快速入门文章对你有所帮助！开始你的 SQLAlchemy 实践之旅吧！

发表评论

您的邮箱地址不会被公开。 必填项已用 * 标注

滚动至顶部