---

SQLAlchemy 入门指南：从核心到 ORM 的全面解析

数据库是现代应用程序的基石。无论是网站、桌面应用还是移动后端，几乎都离不开数据的存储和管理。Python 语言提供了多种与数据库交互的方式，其中 SQLAlchemy 是最强大、最灵活、也是最受 Python 开发者欢迎的数据库工具包和对象关系映射器（ORM）。

对于初学者来说，SQLAlchemy 的功能丰富有时也意味着其学习曲线相对陡峭。但一旦掌握，你会发现它能极大地提高数据库操作的效率和安全性。本篇文章将带你一步步深入 SQLAlchemy 的世界，从其基本概念、安装，到核心功能（Core）和高级功能（ORM）的使用，为你铺平学习之路。

我们将详细讲解：

1. 为什么选择 SQLAlchemy？
2. SQLAlchemy 的核心概念：Core 与 ORM
3. 安装与连接数据库
4. SQLAlchemy Core：数据库抽象层
   • 元数据（MetaData）和表（Table）的定义
   • 创建表
   • 插入数据
   • 查询数据
   • 更新数据
   • 删除数据
5. SQLAlchemy ORM：对象关系映射
   • 声明式映射（Declarative Base）
   • 定义模型（Model）
   • 理解 Session
   • 使用 Session 进行 CRUD 操作 (创建、读取、更新、删除)
   • 基本关系映射 (一对多)
6. Core 与 ORM 的结合与选择
7. Session 管理的最佳实践
8. 总结与进阶方向

本文力求详细，包含丰富的代码示例，帮助你理解每一个概念。

1. 为什么选择 SQLAlchemy？

在 Python 中与数据库交互，你可以直接使用各种数据库驱动（如 psycopg2 for PostgreSQL, mysql-connector-python for MySQL, sqlite3 built-in for SQLite）执行原始 SQL 语句。这样做的好处是直接、性能高，但缺点也很明显：

• 数据库方言差异： 不同的数据库系统（MySQL, PostgreSQL, SQLite, Oracle, SQL Server 等）有不同的 SQL 方言，代码移植性差。
• 手动 SQL 构建： 需要手动拼接 SQL 字符串，容易出错，也容易遭受 SQL 注入攻击。
• 数据与对象分离： 从数据库查询到的数据通常是元组或字典列表，需要手动转换成 Python 对象，反之亦然，非常繁琐。

SQLAlchemy 解决了这些问题。它是一个完整的数据库工具包，主要提供两个功能层：

• SQLAlchemy Core: 提供了一个灵活的 SQL 抽象层，允许你使用 Python 代码构建 SQL 语句，而无需关心具体的数据库方言。它负责将 Python 代码转换成对应数据库的 SQL 语句并执行。
• SQLAlchemy ORM: 在 Core 的基础上，提供了强大的对象关系映射功能。它允许你将 Python 类映射到数据库表，将对象实例映射到表中的行，将对象的属性映射到表的列。通过操作 Python 对象，即可完成数据库的 CRUD 操作，极大地提高了开发效率和代码的可读性。

选择 SQLAlchemy 的理由：

• 跨数据库兼容性： 强大的抽象层屏蔽了数据库方言差异。
• 安全性： 内置参数绑定机制，有效防止 SQL 注入。
• 提高开发效率： ORM 让你使用面向对象的方式操作数据库。
• 灵活性： 你可以自由选择使用 Core 还是 ORM，甚至混合使用，以适应不同的场景需求（例如，复杂查询可能更适合使用 Core）。
• 成熟稳定： SQLAlchemy 是一个经过长期发展和广泛应用的库，拥有活跃的社区和丰富的文档。

2. SQLAlchemy 的核心概念：Core 与 ORM

理解 Core 和 ORM 的区别与联系是学习 SQLAlchemy 的关键。

• SQLAlchemy Core: 就像一个高级的数据库驱动。它抽象了数据库连接、事务、SQL 表达式构建、结果集处理等底层细节。你可以用 Python 代码写出类似 SQL 的结构，SQLAlchemy 会将其翻译成具体的数据库 SQL。它更接近 SQL 本身，但提供了更安全、更具可移植性的方式来构建和执行 SQL。适用于需要精细控制 SQL、追求极致性能或处理复杂、非ORM友好查询的场景。
• SQLAlchemy ORM: 建立在 Core 之上。它将数据库表、行、列与 Python 类、对象、属性关联起来。你定义 Python 类来代表数据库中的实体（如 User, Product），然后使用 Session 对象来管理这些对象的持久化（保存、加载、删除）。ORM 让你感觉像是在操作普通的 Python 对象一样操作数据库。适用于大多数业务场景，可以显著简化代码。

你可以将 Core 视为 SQLAlchemy 的“汇编语言”或“C 语言”，而 ORM 则是其“高级语言”。大多数日常开发会主要使用 ORM，但在需要时，可以直接退回到 Core 层面进行操作，这是 SQLAlchemy 强大灵活性的体现。

3. 安装与连接数据库

首先，你需要安装 SQLAlchemy。通常使用 pip：

bash
pip install sqlalchemy

SQLAlchemy 是一个数据库 工具包，它本身不包含数据库驱动。你需要根据你使用的数据库安装相应的驱动。常用的驱动有：

• SQLite: Python 内置，无需额外安装。
• PostgreSQL: pip install psycopg2
• MySQL: pip install mysql-connector-python 或 pip install PyMySQL
• SQL Server: pip install pyodbc 或 pip install pymssql

在本指南中，我们将主要使用易于上手的 SQLite 数据库，因为它无需额外安装驱动和配置服务器，非常适合入门。

连接数据库

使用 create_engine() 函数来建立与数据库的连接。这是 SQLAlchemy 的入口点。

“`python
from sqlalchemy import create_engine

SQLite 数据库文件路径

file.db 将在当前目录下创建

DATABASE_URL = “sqlite:///file.db”

创建数据库引擎

echo=True 会打印出 SQLAlchemy 生成的所有 SQL 语句，非常有助于学习和调试

engine = create_engine(DATABASE_URL, echo=True)

现在 engine 对象已经创建，你可以用它来与数据库交互了

print(f”数据库引擎已创建: {engine}”)
“`

create_engine() 的第一个参数是数据库连接字符串，格式通常是 dialect+driver://user:password@host:port/database。对于 SQLite 文件数据库，格式简化为 sqlite:///path/to/database/file。三个斜杠 /// 表示绝对路径或当前目录下的相对路径。如果只用两个斜杠 // 后面跟着文件名，表示相对路径。在 Windows 上，路径可能需要多一个斜杠，如 sqlite:///C:/path/to/file.db。对于内存中的 SQLite 数据库（数据不会持久化），连接字符串是 sqlite:///:memory:。

echo=True 是一个非常有用的参数，它会让 SQLAlchemy 记录并打印出它执行的每一条 SQL 语句。这能帮助你理解 SQLAlchemy 如何与数据库交互，以及它在背后做了什么。在生产环境中通常会设置为 False。

engine 对象代表了与数据库的连接池。当你需要执行数据库操作时，可以从 engine 获取连接。

4. SQLAlchemy Core：数据库抽象层

SQLAlchemy Core 允许我们以编程的方式定义数据库结构和构建 SQL 语句。

4.1 元数据（MetaData）和表（Table）的定义

MetaData 对象是关于整个数据库结构的容器。它包含了所有表（Table）、序列等对象的定义。

Table 对象代表数据库中的一个表。它通过名称、关联的 MetaData 对象以及一系列 Column 对象来定义。

Column 对象代表表中的一列。它需要指定列名、数据类型（如 Integer, String, Text, DateTime, Boolean 等）以及可选的约束（如 primary_key=True, ForeignKey, default, nullable=False）。

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

假设 engine 已经创建，如上面所示

DATABASE_URL = “sqlite:///:memory:” # 使用内存数据库方便示例
engine = create_engine(DATABASE_URL, echo=True)

1. 定义 MetaData

metadata = MetaData()

2. 定义 Table

定义 users 表

users_table = Table(
“users”, metadata,
Column(“id”, Integer, primary_key=True), # 主键，会自动生成
Column(“name”, String, nullable=False), # 非空字符串
Column(“fullname”, String),
Column(“age”, Integer)
)

定义 addresses 表

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, nullable=False)
)

print(“表定义完成，但尚未创建数据库表。”)
“`

在上面的代码中，我们定义了 users 和 addresses 两个表。users 表有 id, name, fullname, age 四列，其中 id 是主键。addresses 表有 id, user_id, email_address 三列，其中 id 是主键，user_id 是外键，关联到 users 表的 id 列。

4.2 创建表

定义好 MetaData 和 Table 对象后，可以使用 metadata.create_all(engine) 方法来在数据库中创建这些表。SQLAlchemy 会检查数据库中是否已存在同名表，如果不存在，则根据定义执行 CREATE TABLE 语句。

“`python

使用 engine 连接数据库，创建所有在 metadata 中定义的表

metadata.create_all(engine)

print(“数据库表已创建（如果不存在的话）。”)

查看 echo=True 输出，可以看到 CREATE TABLE 语句

“`

4.3 插入数据

使用 Core 插入数据，需要构建 insert() 表达式，并通过连接对象执行。

“`python

导入 insert 函数

from sqlalchemy import insert

获取一个数据库连接

with 语句确保连接在使用后被正确关闭

with engine.connect() as connection:
# 构建 insert 表达式
# 插入单行数据
insert_stmt_single = insert(users_table).values(name=”spongebob”, fullname=”Spongebob Squarepants”)

# 执行插入操作
result_single = connection.execute(insert_stmt_single)
# 对于大多数数据库，插入单行数据后，可以通过 result.inserted_primary_key 获取新插入行的主键
# 注意：SQLite 3.35.0+ 返回的是列表，早期版本直接是值
inserted_user_id_single = result_single.inserted_primary_key[0] if result_single.inserted_primary_key else None
print(f"插入单行用户成功，新用户 ID: {inserted_user_id_single}")

# 插入多行数据
insert_stmt_multiple = insert(users_table)
users_to_insert = [
    {"name": "patrick", "fullname": "Patrick Star"},
    {"name": "sandy", "fullname": "Sandy Cheeks"},
]
result_multiple = connection.execute(insert_stmt_multiple, users_to_insert)
print(f"插入多行用户成功。")

# 插入地址数据，关联到用户
insert_addresses_stmt = insert(addresses_table)
addresses_to_insert = [
    {"user_id": inserted_user_id_single, "email_address": "[email protected]"},
    {"user_id": inserted_user_id_single, "email_address": "[email protected]"},
]
connection.execute(insert_addresses_stmt, addresses_to_insert)
print("插入地址数据成功。")

# 提交事务 (Core 默认是autocommit=False，需要手动commit)
# 或者使用 connection.commit()
connection.commit() # 在with语句中通常需要手动commit，除非engine配置了autocommit=True

print(“数据插入完成。”)
“`

使用 with engine.connect() as connection: 是推荐的方式，它会自动获取连接并在退出 with 块时释放连接。在 Core 中，通常需要显式地调用 connection.commit() 来保存更改。

execute() 方法的第二个参数可以是一个字典（单行）或字典列表（多行），SQLAlchemy 会自动进行参数绑定，这比手动格式化字符串安全得多。

4.4 查询数据

使用 Core 查询数据需要构建 select() 表达式，执行它，然后处理结果集。

“`python

导入 select 函数

from sqlalchemy import select, or_

with engine.connect() as connection:
# 查询所有用户
select_all_users_stmt = select(users_table) # select * from users
result_all_users = connection.execute(select_all_users_stmt)

print("\n--- 查询所有用户 ---")
for row in result_all_users:
    print(row) # row 是一个 RowProxy 对象，类似元组和字典

# 查询特定列
select_names_stmt = select(users_table.c.name, users_table.c.fullname) # select name, fullname from users
result_names = connection.execute(select_names_stmt)

print("\n--- 查询用户姓名和全名 ---")
for name, fullname in result_names: # 可以像解包元组一样获取列值
    print(f"Name: {name}, Fullname: {fullname}")

# 带条件的查询 (WHERE 子句)
# 查找 name 是 'spongebob' 的用户
select_spongebob_stmt = select(users_table).where(users_table.c.name == "spongebob")
result_spongebob = connection.execute(select_spongebob_stmt)

print("\n--- 查询名字为 'spongebob' 的用户 ---")
for row in result_spongebob:
    print(row)

# 复杂的条件 (AND, OR)
# 查找 id 大于 1 且 name 为 'patrick' 或 'sandy' 的用户
select_complex_stmt = select(users_table).where(
    (users_table.c.id > 1) & # AND 操作符，使用 &
    or_( # OR 操作符，使用 or_ 函数
        users_table.c.name == "patrick",
        users_table.c.name == "sandy"
    )
)
result_complex = connection.execute(select_complex_stmt)

print("\n--- 查询 id > 1 且 (name = 'patrick' 或 name = 'sandy') 的用户 ---")
for row in result_complex:
    print(row)

# 带限制和偏移的查询 (LIMIT, OFFSET)
select_limited_stmt = select(users_table).limit(1).offset(1) # LIMIT 1 OFFSET 1
result_limited = connection.execute(select_limited_stmt)

print("\n--- 查询 LIMIT 1 OFFSET 1 ---")
for row in result_limited:
    print(row)

# 带排序的查询 (ORDER BY)
select_ordered_stmt = select(users_table).order_by(users_table.c.name.desc()) # ORDER BY name DESC
result_ordered = connection.execute(select_ordered_stmt)

print("\n--- 查询所有用户按姓名倒序 ---")
for row in result_ordered:
    print(row)

# 连接查询 (JOIN)
# 查询所有用户及其邮箱地址
select_join_stmt = select(users_table, addresses_table).\
    join(addresses_table, users_table.c.id == addresses_table.c.user_id)
result_join = connection.execute(select_join_stmt)

print("\n--- 连接 users 和 addresses 表查询 ---")
for row in result_join:
    print(row) # 结果包含来自两个表的所有列
    # 也可以按表名或列名访问
    # print(f"User Name: {row[users_table.c.name]}, Email: {row[addresses_table.c.email_address]}")

# 更简洁的 join 查询
select_join_simplified_stmt = select(users_table.c.name, addresses_table.c.email_address).\
    select_from(users_table.join(addresses_table)) # select_from 指定 FROM 和 JOIN 子句
result_join_simplified = connection.execute(select_join_simplified_stmt)
print("\n--- 简化连接查询只获取特定列 ---")
for user_name, email in result_join_simplified:
     print(f"User: {user_name}, Email: {email}")

print(“数据查询完成。”)
“`

查询结果是一个可迭代的 ResultProxy 对象，每一行是一个 RowProxy 对象。RowProxy 行为类似于元组和字典，你可以通过索引或列名（使用 table.c.column_name）访问列的值。

select() 函数构建了 SELECT 语句。where() 方法用于添加 WHERE 子句。逻辑运算符 AND 使用 & 符号，OR 使用 or_() 函数。limit() 和 offset() 分别对应 SQL 的 LIMIT 和 OFFSET。order_by() 用于添加 ORDER BY 子句，.desc() 可以实现倒序。join() 用于执行 JOIN 操作。

4.5 更新数据

使用 Core 更新数据需要构建 update() 表达式，并通过连接对象执行。

“`python

导入 update 函数

from sqlalchemy import update

with engine.connect() as connection:
# 构建 update 表达式
# 将名字为 ‘spongebob’ 的用户的全名更新为 ‘S.B. Squarepants’
update_stmt = update(users_table).where(users_table.c.name == “spongebob”).values(fullname=”S.B. Squarepants”)

# 执行更新操作
result_update = connection.execute(update_stmt)

# 查看影响的行数
print(f"\n更新操作影响了 {result_update.rowcount} 行。")

# 提交事务
connection.commit()

print(“数据更新完成。”)

验证更新结果

with engine.connect() as connection:
select_spongebob_stmt = select(users_table).where(users_table.c.name == “spongebob”)
updated_user = connection.execute(select_spongebob_stmt).fetchone() # fetchone() 获取第一行结果或 None
print(f”更新后的用户数据: {updated_user}”)
“`

update() 函数构建了 UPDATE 语句。where() 方法指定更新条件，values() 方法指定要更新的列及其新值。execute() 返回的 ResultProxy 对象有 rowcount 属性，表示受影响的行数。

4.6 删除数据

使用 Core 删除数据需要构建 delete() 表达式，并通过连接对象执行。

“`python

导入 delete 函数

from sqlalchemy import delete

with engine.connect() as connection:
# 构建 delete 表达式
# 删除名字为 ‘patrick’ 的用户
delete_stmt = delete(users_table).where(users_table.c.name == “patrick”)

# 执行删除操作
result_delete = connection.execute(delete_stmt)

# 查看影响的行数
print(f"\n删除操作影响了 {result_delete.rowcount} 行。")

# 提交事务
connection.commit()

print(“数据删除完成。”)

验证删除结果

with engine.connect() as connection:
select_patrick_stmt = select(users_table).where(users_table.c.name == “patrick”)
patrick_user = connection.execute(select_patrick_stmt).fetchone()
print(f”删除后的用户数据 (Patrick): {patrick_user}”) # 应该是 None
“`

delete() 函数构建了 DELETE 语句。where() 方法指定删除条件。同样，rowcount 属性表示受影响的行数。

Core 小结：

SQLAlchemy Core 提供了一种强大且类型安全的方式来构建和执行 SQL 语句，而无需编写原始 SQL 字符串。它适用于需要底层控制或处理复杂查询的场景。虽然相比 ORM 代码量稍多，但它更接近数据库操作的本质。

5. SQLAlchemy ORM：对象关系映射

SQLAlchemy ORM 是建立在 Core 之上的高级抽象层。它允许你使用 Python 类来定义数据库模型，通过操作 Python 对象来与数据库交互。这是 SQLAlchemy 最常用也是最强大的功能。

5.1 声明式映射（Declarative Base）

ORM 的核心是将 Python 类映射到数据库表。SQLAlchemy 提供了多种映射方式，其中“声明式”（Declarative）是最常用和推荐的方式。它允许你在同一个类定义中同时定义表结构和类属性到列的映射。

“`python
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy.ext.declarative import declarative_base # 在新版本中推荐使用 sqlalchemy.orm.declarative_base

在 SQLAlchemy 1.4 及更高版本，推荐使用 sqlalchemy.orm.declarative_base

try:
from sqlalchemy.orm import declarative_base
except ImportError:
from sqlalchemy.ext.declarative import declarative_base # 兼容旧版本

创建数据库引擎

DATABASE_URL = “sqlite:///:memory:” # 继续使用内存数据库
engine = create_engine(DATABASE_URL, echo=True)

1. 创建声明式基类

这个基类是所有 ORM 模型的父类。它包含了 MetaData 和一些映射功能。

Base = declarative_base()

print(“声明式基类已创建。”)
“`

declarative_base() 函数返回一个基类 Base。你的所有 ORM 模型类都需要继承自这个 Base 类。

5.2 定义模型（Model）

继承 Base 类来定义你的数据库模型。每个模型类对应一个数据库表。

“`python

2. 定义 ORM 模型类

class User(Base):
tablename = ‘users’ # 映射到的数据库表名

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

# 定义与 Address 的关系：一个用户可以有多个地址
# relationship('Address', backref='user') 表示 User 类通过 'addresses' 属性引用 Address 对象列表
# Address 类通过 'user' 属性引用 User 对象
addresses = relationship("Address", back_populates="user", cascade="all, delete, delete-orphan")

def __repr__(self):
    return f"<User(id={self.id}, name='{self.name}', fullname='{self.fullname}', age={self.age})>"

class Address(Base):
tablename = ‘addresses’

id = Column(Integer, primary_key=True)
user_id = Column(Integer, ForeignKey('users.id'), nullable=False) # 外键关联到 users 表的 id 列
email_address = Column(String, nullable=False)

# 定义与 User 的关系：一个地址属于一个用户
# back_populates='addresses' 指向 User 类中对应的 relationship 属性名
user = relationship("User", back_populates="addresses")

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

print(“ORM 模型类已定义。”)
“`

在模型类中：

• __tablename__ 是必需的类属性，指定了该类映射到的数据库表名。
• 类属性（如 id, name, email_address）使用 Column() 函数定义，这与 Core 中的 Column 类似，但直接作为类属性。参数和数据类型也与 Core 相同。
• relationship() 函数用于定义模型之间的关系（如一对多、多对多）。back_populates 参数用于建立双向关系，使其更容易从关系的两端导航。cascade 参数用于级联操作，例如 all, delete, delete-orphan 意味着当删除一个 User 对象时，其关联的 Address 对象也会被删除。

定义完模型后，可以使用 Base.metadata.create_all(engine) 方法来在数据库中创建对应的表。这与 Core 中的 metadata.create_all() 类似，因为 Base.metadata 就是一个 MetaData 实例。

“`python

3. 创建数据库表

Base.metadata.create_all(engine)
print(“数据库表已通过 ORM 模型定义创建。”)

查看 echo=True 输出，可以看到 CREATE TABLE 语句

“`

5.3 理解 Session

在 ORM 中，你不是直接通过 engine 或 connection 来操作对象，而是通过一个 Session 对象。Session 是 ORM 的核心工作区域，它：

• 代表了一个与数据库的“对话”或“事务”。
• 跟踪你创建、修改、删除的对象（称为“持久化对象”）。
• 负责将对象状态的变化翻译成 SQL 语句并发送到数据库。
• 提供了查询功能，将数据库行转换回 Python 对象。

Session 对象通常使用 sessionmaker 工厂函数来创建。

“`python

4. 创建 Session 工厂

sessionmaker() 返回一个类，每次调用这个类就会创建一个新的 Session 实例

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

创建一个 Session 实例

session = SessionLocal() # 在需要使用时调用

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

sessionmaker 的常用参数：

• bind: 必须指定一个 Engine 对象，Session 将使用它获取连接。
• autocommit: 如果设置为 True，每次调用 add(), delete(), update() 等方法后都会自动提交事务。通常建议设置为 False，手动控制事务提交 (session.commit())，这样可以将多个操作放在一个事务中。
• autoflush: 如果设置为 True (默认值)，在执行查询之前会自动将 Session 中待同步的对象状态刷新到数据库，但不会提交。通常建议设置为 False，除非你清楚其作用。

创建 SessionLocal 工厂后，每次需要与数据库交互时，就创建一个新的 Session 实例。使用 with 语句管理 Session 是最佳实践，可以确保 Session 在使用完毕后被关闭，即使发生异常：

“`python

获取一个 Session 实例

with SessionLocal() as session:
# 在这里进行 ORM 操作
pass # 操作完成后会自动关闭 session

或者手动管理:

session = SessionLocal()

try:

# ORM 操作

session.commit()

except Exception as e:

session.rollback()

raise

finally:

session.close()

“`

5.4 使用 Session 进行 CRUD 操作

现在，我们可以使用 Session 对象来操作我们定义的 ORM 模型对象了。

创建 (Create)

创建 ORM 对象非常简单，就像创建普通 Python 对象一样。然后使用 session.add() 或 session.add_all() 将对象添加到 Session 中。最后调用 session.commit() 将更改保存到数据库。

“`python

插入数据

with SessionLocal() as session:
# 创建 User 对象
new_user = User(name=”spongebob”, fullname=”Spongebob Squarepants”, age=25)
another_user = User(name=”patrick”, fullname=”Patrick Star”, age=30)

# 将对象添加到 session
session.add(new_user)
session.add(another_user)

# 创建关联的 Address 对象
spongebob_address1 = Address(email_address="[email protected]")
spongebob_address2 = Address(email_address="[email protected]")
patrick_address = Address(email_address="[email protected]")

# 建立对象间的关系
new_user.addresses.append(spongebob_address1) # 直接将 Address 对象添加到 User 对象的 addresses 列表
new_user.addresses.append(spongebob_address2)
patrick_address.user = another_user # 直接设置 Address 对象的 user 属性

# 添加 Address 对象到 session
# 注意：如果通过关系添加到 User 对象，并且 User 对象已经 add 到 session，Address 对象也会自动被 session 跟踪，无需单独 add
# 但为了清晰，我们也可以显式 add
session.add_all([spongebob_address1, spongebob_address2, patrick_address])

print("\n添加对象到 session...")
# 在 commit 之前，对象可能还没有 ID (取决于数据库方言和配置)
print(f"new_user (before commit): {new_user}")

# 提交 session，将更改保存到数据库
session.commit()
print("Session 提交成功，数据已保存到数据库。")

# commit 后，对象会自动拥有数据库生成的主键 ID
print(f"new_user (after commit): {new_user}")

# 如果需要获取刚插入对象的 ID，可以在 commit 后访问其主键属性
inserted_spongebob_id = new_user.id
inserted_patrick_id = another_user.id
print(f"新插入的用户 ID: Spongebob={inserted_spongebob_id}, Patrick={inserted_patrick_id}")

# 你也可以使用 session.add_all() 一次添加多个对象
more_users = [
    User(name="sandy", fullname="Sandy Cheeks", age=28),
    User(name="squidward", fullname="Squidward Tentacles", age=40),
]
session.add_all(more_users)
session.commit()
print(f"又插入了 {len(more_users)} 个用户。")

print(“ORM 数据插入完成。”)
“`

读取 (Read / Query)

使用 session.query() 方法来构建查询。你可以查询模型类，然后使用各种方法（如 filter(), order_by(), limit(), offset() 等）细化查询条件，最后使用执行方法（如 all(), first(), one(), get(), scalar() 等）获取结果。

“`python

查询数据

with SessionLocal() as session:
print(“\n— 查询所有用户 —“)
users = session.query(User).all() # 查询所有 User 对象，返回列表
for user in users:
print(user)
# 通过关系访问关联对象
print(f” Addresses: {user.addresses}”) # 访问 related Address objects

print("\n--- 查询名字为 'spongebob' 的用户 ---")
# filter() 使用 Python 表达式风格的条件
spongebob = session.query(User).filter(User.name == 'spongebob').first() # 返回第一个匹配的对象或 None
print(spongebob)
if spongebob:
    print(f"  Spongebob 的地址: {spongebob.addresses}")

print("\n--- 查询年龄大于等于 30 的用户 ---")
# filter() 可以链式调用，表示 AND 关系
old_users = session.query(User).filter(User.age >= 30).all()
for user in old_users:
    print(user)

print("\n--- 复杂的条件查询 (OR) ---")
# 使用 sqlalchemy.or_ 函数
from sqlalchemy import or_
patrick_or_sandy = session.query(User).filter(or_(User.name == 'patrick', User.name == 'sandy')).all()
for user in patrick_or_sandy:
    print(user)

print("\n--- 按年龄倒序查询，限制 2 个 ---")
ordered_limited_users = session.query(User).\
    order_by(User.age.desc()).\
    limit(2).\
    all()
for user in ordered_limited_users:
    print(user)

print("\n--- 按主键查询 (get) ---")
# get() 是根据主键获取对象的快捷方式，如果对象已经在 Session 中，直接返回，否则查询数据库
user_by_id_1 = session.get(User, 1) # 根据主键 1 获取 User 对象
print(f"通过 ID 1 获取的用户: {user_by_id_1}")
user_by_id_99 = session.get(User, 99)
print(f"通过 ID 99 获取的用户: {user_by_id_99}") # 应该返回 None

print("\n--- 连接查询 (使用 join) ---")
# 查询拥有 email 地址的用户姓名和邮箱
users_with_email = session.query(User.name, Address.email_address).\
    join(Address).\
    filter(User.name == 'spongebob').\
    all()
print(f"Spongebob 的邮箱地址列表: {users_with_email}")

# 查询每个用户及其地址数量
from sqlalchemy import func # 导入聚合函数
user_address_counts = session.query(User.name, func.count(Address.id)).\
    outerjoin(Address).\
    group_by(User.id).\
    all()
print("\n--- 用户及其地址数量 ---")
for name, count in user_address_counts:
    print(f"User: {name}, Addresses Count: {count}")

print(“ORM 数据查询完成。”)
“`

query() 方法非常灵活。它接受模型类或模型属性作为参数。filter() 相当于 SQL 的 WHERE 子句。all() 返回所有匹配的对象列表，first() 返回第一个匹配的对象（如果没有则返回 None），one() 返回一个且只有一个匹配对象（否则抛出异常），get() 根据主键获取对象。join() 和 Core 中的类似，用于关联查询。func 可以用来使用 SQL 聚合函数。

更新 (Update)

更新 ORM 对象很简单：先通过查询获取对象，然后修改其属性，最后提交 Session。Session 会自动跟踪对象的修改，并在提交时生成 UPDATE 语句。

“`python

更新数据

with SessionLocal() as session:
# 获取要更新的用户
spongebob_user = session.query(User).filter(User.name == ‘spongebob’).first()

if spongebob_user:
    print(f"\n更新前: {spongebob_user}")
    # 修改对象属性
    spongebob_user.age = 26
    spongebob_user.fullname = "SpongeBob SquarePants (Updated)"

    # 修改关联对象属性
    if spongebob_user.addresses:
         spongebob_user.addresses[0].email_address = "[email protected]"

    print(f"更新后 (内存中): {spongebob_user}")

    # 提交 session，将更改保存到数据库
    session.commit()
    print("Session 提交成功，更新已保存到数据库。")

验证更新

with SessionLocal() as session:
updated_spongebob = session.get(User, spongebob_user.id)
print(f”从数据库重新加载后: {updated_spongebob}”)
print(f”其地址: {updated_spongebob.addresses}”)

print(“ORM 数据更新完成。”)
“`

删除 (Delete)

删除 ORM 对象也类似：先通过查询获取对象，然后使用 session.delete() 将对象标记为待删除，最后提交 Session。

“`python

删除数据

with SessionLocal() as session:
# 获取要删除的用户 (例如 Patrick)
patrick_user = session.query(User).filter(User.name == ‘patrick’).first()

if patrick_user:
    print(f"\n准备删除用户: {patrick_user}")
    # 删除对象
    session.delete(patrick_user)
    print("对象已标记为删除。")

    # 提交 session，将更改保存到数据库
    session.commit()
    print("Session 提交成功，用户已从数据库删除。")
    # 因为我们在 User 模型中设置了 cascade="all, delete, delete-orphan"，
    # 关联的 Address 对象也会被删除。

验证删除

with SessionLocal() as session:
deleted_patrick = session.query(User).filter(User.name == ‘patrick’).first()
print(f”查询删除的用户: {deleted_patrick}”) # 应该返回 None
# 查询 Patrick 的地址，应该也已被删除
patrick_addresses = session.query(Address).filter(Address.email_address == ‘[email protected]’).all()
print(f”查询 Patrick 的地址: {patrick_addresses}”) # 应该返回空列表 []

print(“ORM 数据删除完成。”)
“`

5.5 基本关系映射 (一对多)

在上面的例子中，我们已经初步展示了如何使用 relationship() 定义一对多的关系，以及如何通过模型对象的属性来访问关联的对象 (user.addresses 和 address.user)。

• 在“一”方 (User 类) 定义 relationship()，指向“多”方 (Address 类)。例如：addresses = relationship("Address", ...)。
• 在“多”方 (Address 类) 定义 relationship()，指向“一”方 (User 类)，并且通常包含一个 ForeignKey 列指向“一”方的主键。例如：user_id = Column(Integer, ForeignKey('users.id')), user = relationship("User", ...)。
• 使用 back_populates 或 backref 来建立双向关系。back_populates (推荐) 需要在关系的两端都定义，并互相引用对方的属性名。backref 只需要在一端定义，SQLAlchemy 会自动在另一端创建反向引用的属性。

通过关系属性访问关联对象时，SQLAlchemy 会懒加载（Lazy Loading）数据，即在第一次访问该属性时才执行相应的 SQL 查询去数据库中获取关联数据。这是一种性能优化策略。

6. Core 与 ORM 的结合与选择

SQLAlchemy 的强大之处在于你可以根据需要选择使用 Core 还是 ORM，甚至将它们结合起来。

• 何时使用 ORM:
  • 大多数业务逻辑操作，涉及对象的创建、修改、删除。
  • 简单的查询，特别是通过关系导航的查询。
  • 当你希望代码更具可读性，以面向对象的方式思考数据库操作时。
• 何时使用 Core:
  • 执行数据库管理任务，如创建/删除表、修改表结构（虽然大部分由 ORM 映射生成）。
  • 执行批量插入/更新/删除操作，特别是从其他数据源加载大量数据时，Core 通常比 ORM 的 add/commit 组合性能更高。
  • 执行非常复杂或高度优化的查询，ORM 的自动 SQL 生成可能不够理想，或者你希望精确控制生成的 SQL。
  • 使用数据库特有的函数或特性，ORM 可能没有直接的抽象。

你甚至可以在 ORM 的 Session 中执行 Core 表达式：

“`python
from sqlalchemy import select

with SessionLocal() as session:
# 使用 Core select 表达式查询，并在 ORM session 中执行
core_select_stmt = select(users_table.c.name).where(users_table.c.age > 30)
# 使用 session.execute() 执行 Core 表达式
result = session.execute(core_select_stmt)

print("\n--- 在 ORM Session 中执行 Core 查询 ---")
for row in result:
    print(row.name) # 结果行类似 Core 的 RowProxy

print(“在 ORM Session 中执行 Core 查询完成。”)
“`

7. Session 管理的最佳实践

正确的 Session 管理是使用 SQLAlchemy ORM 的关键。一个常见的错误是将 Session 视为全局或长期存在的对象。Session 应该是短暂的，与特定的事务或请求生命周期绑定。

推荐的最佳实践：

• 一个请求一个 Session： 在 Web 应用程序中，每个传入的请求都应该有一个新的 Session。请求开始时创建 Session，请求结束时提交或回滚事务并关闭 Session。
• 使用 with 语句： 利用 Python 的上下文管理器 (with SessionLocal() as session:) 来自动处理 Session 的关闭和异常时的回滚。
• 避免在函数之间传递 Session： 尽量让 Session 的作用域限制在一个函数或一个逻辑单元内。如果需要跨函数访问，考虑使用依赖注入等模式。
• 不要在多线程或多协程之间共享同一个 Session： Session 不是线程安全的。每个线程/协程应该有自己的 Session 实例。sessionmaker 就是为了方便创建新的 Session 实例而设计的。

8. 总结与进阶方向

恭喜你！现在你已经了解了 SQLAlchemy 的基本概念，包括 Core 和 ORM 的使用，如何连接数据库，如何定义模型和表，以及如何执行基本的 CRUD 操作。

回顾关键点：

• create_engine(): 建立数据库连接。
• Core: MetaData, Table, Column, insert(), select(), update(), delete(), engine.connect(), connection.execute(), connection.commit().
• ORM: declarative_base(), 模型类 (__tablename__, Column, relationship), sessionmaker(), Session, session.add(), session.add_all(), session.delete(), session.query(), session.commit(), session.rollback(), session.close().
• Session 是 ORM 的工作单元，管理对象状态和事务。
• with SessionLocal() as session: 是推荐的 Session 管理方式。

进阶方向：

这仅仅是 SQLAlchemy 的冰山一角。要更深入地掌握它，你可以继续学习：

• 更高级的关系映射： 多对多关系、自引用关系、复合主键等。
• 更复杂的查询： 子查询、联结类型（左外连接等）、过滤操作符（like, in_, is null 等）、Having 子句、CTE (Common Table Expressions)。
• 数据类型： 枚举类型、JSON 类型、数组类型等数据库特有类型的使用。
• 事务管理： 隔离级别、保存点。
• 性能优化： 延迟加载（Lazy Loading）的策略（joined, subquery, noload 等）、 eager loading、核心层面的优化。
• 事件系统： 在对象或 Session 事件发生时执行自定义逻辑。
• 迁移工具： 使用 Alembic 等工具管理数据库模式的变更。
• 连接池配置： 调整 create_engine 参数以优化连接管理。
• 错误处理： 了解 SQLAlchemy 抛出的异常类型及其处理方式。

SQLAlchemy 文档非常详尽，是进一步学习的最佳资源。通过实践和不断探索，你将能够充分发挥 SQLAlchemy 的强大功能，更高效地进行数据库开发。

现在，就开始动手尝试修改上面的代码示例，连接你自己的数据库，创建更复杂的模型，执行更多样的查询吧！祝你学习愉快！

---