SQLAlchemy 介绍与使用：一个全面指南

引言

在现代软件开发中，数据库是几乎所有应用程序的核心。与数据库交互是日常任务，而如何高效、安全、灵活地进行交互则至关重要。Python 作为一门功能强大且用途广泛的编程语言，与数据库的交互方式也多种多样。其中，SQLAlchemy 无疑是 Python 生态系统中最流行、最强大、最灵活的数据库工具包和对象关系映射器（ORM）。

SQLAlchemy 不仅仅是一个简单的 ORM，它提供了一个完整的企业级持久层解决方案。它既提供了低级别的数据库抽象，允许开发者编写类似 SQL 的语句，也提供了高级别的 ORM，将 Python 对象映射到数据库表，极大地简化了数据操作。

本文将深入探讨 SQLAlchemy 的核心概念、两大部分（SQLAlchemy Core 和 SQLAlchemy ORM）以及它们的具体使用方法，帮助你全面理解并开始使用这个强大的库。

什么是 SQLAlchemy?

SQLAlchemy 是一个 Python SQL 工具包和对象关系映射器。它的设计哲学是提供数据库的全部力量和灵活性，同时提供易于使用且高效的抽象。它不是试图隐藏 SQL 的所有方面，而是提供一种 Pythonic 的方式来构建和执行 SQL 语句，或者将 Python 对象映射到数据库结构。

SQLAlchemy 主要由两大部分组成：

SQLAlchemy Core: 这是一个数据库抽象层，提供了连接池、事务管理、SQL 表达式构建器（SQL Expression Language）等功能。它允许开发者以 Python 的方式构造 SQL 语句，而无需编写字符串形式的 SQL。这提供了更高的安全性（防止 SQL 注入）和更好的可读性。你可以把它看作是一个高级的 DB-API 接口。
SQLAlchemy ORM: 这是建立在 Core 之上的高级层。它提供了一种将 Python 类和对象与数据库表和行进行映射的方式。你可以像操作 Python 对象一样操作数据库数据，ORM 会自动将这些操作翻译成 SQL 语句并执行。ORM 提供了对象身份映射（Identity Mapping）、关系映射（Relationship Mapping）、会话管理等功能，极大地简化了复杂数据模型的操作。

SQLAlchemy 支持广泛的数据库后端，包括 PostgreSQL, MySQL, SQLite, Oracle, Microsoft SQL Server, Firebird, Sybase 等，通过相应的数据库驱动程序（DB-API 实现）进行连接。

安装 SQLAlchemy

安装 SQLAlchemy 非常简单，使用 pip 即可：

bash pip install sqlalchemy

除了 SQLAlchemy 本身，你还需要安装你使用的数据库对应的 Python 驱动。例如：

PostgreSQL: pip install psycopg2-binary
MySQL: pip install mysql-connector-python 或 pip install PyMySQL
SQLite: SQLite 是 Python 标准库的一部分，通常不需要额外安装驱动。

SQLAlchemy Core：深入数据库抽象层

SQLAlchemy Core 是 SQLAlchemy 的基础，它提供了构建 SQL 表达式和与数据库进行低级交互的能力。即使你主要使用 ORM，理解 Core 的概念也是非常有益的。

Core 的主要组件包括：

Engine (引擎): 这是与特定数据库建立连接的入口。它管理着连接池和方言（dialect），负责与数据库驱动程序交互。
MetaData (元数据): 这是一个容器对象，用于存储关于数据库模式的信息，例如表（Table）、列（Column）等。它就像数据库模式在内存中的表示。
Table (表): 代表数据库中的一个表。
Column (列): 代表表中的一个列。
SQL Expression Language: 用于构造 SQL 语句的 Python API。

1. Engine：连接数据库

使用 create_engine() 函数创建一个 Engine 实例，这是与数据库通信的起点。连接 URL 的格式通常是 dialect+driver://user:password@host:port/database。

示例：连接到不同的数据库

“`python
from sqlalchemy import create_engine

SQLite (内存数据库)

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

SQLite (文件数据库)

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

PostgreSQL

engine = create_engine(“postgresql://user:password@host:port/database”)

MySQL

engine = create_engine(“mysql+mysqlconnector://user:password@host:port/database”)

或者使用 PyMySQL

engine = create_engine(“mysql+pymysql://user:password@host:port/database”)

print(f”Engine created for {engine.url.drivername} dialect.”)
“`

create_engine 还有许多参数可以配置连接池、性能等，例如 echo=True 会打印出 SQLAlchemy 执行的每一条 SQL 语句，对于调试非常有用。

2. MetaData, Table, Column：定义数据库结构

在 Core 中，我们使用 MetaData 对象来注册和管理表定义。Table 对象定义了一个具体的表，Column 对象定义了表中的列。

示例：定义一个简单的表

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

创建 MetaData 对象

metadata = MetaData()

定义一个 users 表

users_table = Table(
“users”, # 表名
metadata,
Column(“id”, Integer, primary_key=True), # 主键列
Column(“name”, String(50), nullable=False), # 非空字符串列
Column(“fullname”, String(100)), # 可空字符串列
)

定义一个 addresses 表，包含外键关联 users 表

addresses_table = Table(
“addresses”,
metadata,
Column(“id”, Integer, primary_key=True),
Column(“user_id”, ForeignKey(“users.id”), nullable=False), # 外键列
Column(“email_address”, String(100), nullable=False),
)

print(“Tables defined in MetaData.”)
“`

Table 的第一个参数是表名，第二个参数是所属的 MetaData 对象。
Column 的第一个参数是列名，第二个参数是数据类型（如 Integer, String, Boolean, Date, Float 等，这些都是 SQLAlchemy 提供的通用类型，SQLAlchemy 会将其映射到具体数据库的类型），后面的参数是列的约束或属性（如 primary_key, nullable, default, unique, index, ForeignKey 等）。
ForeignKey 用于定义外键关系，参数格式是 "关联表名.关联列名"。

3. 创建表结构

定义好 MetaData 和其中的表后，可以使用 metadata.create_all() 方法根据定义创建数据库表。

示例：创建表

“`python

假设 engine 已经创建

metadata.create_all(engine)

print(“Tables created in the database.”)

“`

注意: create_all 会检查表是否存在，如果不存在则创建。如果表已存在且结构不一致，它不会修改现有表。对于数据库模式的修改（如添加列、修改列类型等），通常需要使用数据库迁移工具，如 Alembic（专门为 SQLAlchemy 设计）。

4. SQLAlchemy Core CRUD 操作 (SQL Expression Language)

一旦定义了表结构并建立了连接，就可以使用 SQL Expression Language 来执行 CRUD (Create, Read, Update, Delete) 操作了。这通过构建表达式对象并使用 Engine 的连接（Connection）来执行实现。

获取连接通常使用 engine.connect()。为了确保连接正确关闭，推荐使用 Python 的上下文管理器 with engine.connect() as connection:。

示例：插入数据

“`python
from sqlalchemy import insert

构造 insert 语句

insert_users = insert(users_table).values(
name=”spongebob”, fullname=”Spongebob Squarepants”
)

使用连接执行语句

with engine.connect() as connection:
result = connection.execute(insert_users)
# 对于支持返回主键的数据库，可以获取插入行的主键
inserted_user_id = result.inserted_primary_key[0]
connection.commit() # 提交事务
print(f”Inserted user with ID: {inserted_user_id}”)

插入多条数据

insert_addresses = insert(addresses_table)
values_list = [
{“user_id”: inserted_user_id, “email_address”: “[email protected]”},
{“user_id”: inserted_user_id, “email_address”: “[email protected]”},
]

with engine.connect() as connection:
connection.execute(insert_addresses, values_list) # execute 可以接受一个列表执行多次插入
connection.commit()
print(f”Inserted {len(values_list)} addresses.”)
“`

insert(table) 创建一个插入表达式。
.values() 指定要插入的列和值，可以是一个字典或字典列表。
connection.execute() 执行表达式。
result.inserted_primary_key 返回插入行的主键（如果数据库和驱动支持）。
使用 connection.commit() 提交事务，connection.rollback() 回滚事务。

示例：查询数据 (Select)

“`python
from sqlalchemy import select, and_, or_

构造 select 语句

select_users = select(users_table) # 查询 users_table 的所有列

with engine.connect() as connection:
result = connection.execute(select_users)
# 遍历结果集
for row in result:
print(row) # row 是一个 ResultProxy 对象，可以像元组或字典一样访问

print(“-” * 20)

查询特定列

select_names = select(users_table.c.name, users_table.c.fullname) # .c 访问表的列

with engine.connect() as connection:
for row in connection.execute(select_names):
print(f”Name: {row.name}, Fullname: {row.fullname}”) # 通过列名访问

print(“-” * 20)

带条件的查询 (WHERE 子句)

select_spongebob = select(users_table).where(users_table.c.name == “spongebob”)

with engine.connect() as connection:
spongebob_user = connection.execute(select_spongebob).fetchone() # fetchone() 获取第一行
if spongebob_user:
print(f”Found spongebob: {spongebob_user.name} ({spongebob_user.fullname})”)

print(“-” * 20)

多个条件 (AND, OR)

select_complex = select(users_table).where(
and_(users_table.c.name.startswith(“s”), users_table.c.id > 0)
)

或者使用 & 操作符： select(users_table).where((users_table.c.name.startswith(“s”)) & (users_table.c.id > 0))

with engine.connect() as connection:
for row in connection.execute(select_complex):
print(f”Complex query match: {row.name}”)

print(“-” * 20)

联合查询 (JOIN)

select_join = select(users_table.c.name, addresses_table.c.email_address).join(
addresses_table, users_table.c.id == addresses_table.c.user_id
).where(users_table.c.name == “spongebob”)

with engine.connect() as connection:
print(“Joined query results:”)
for row in connection.execute(select_join):
print(f”Name: {row.name}, Email: {row.email_address}”)

“`

select(columns) 创建一个查询表达式，可以指定要查询的列，或表对象（查询所有列）。
.where() 添加 WHERE 子句。条件可以使用 Python 比较运算符（==, !=, >, <, >=, <=, ~ for NOT, is_ for IS NULL, isnot for IS NOT NULL）以及各种表达式方法（如 .startswith(), .contains(), .in_([]), .is_(), .isnot()）。
and_(), or_() 用于组合多个条件。
.join() 用于执行连接操作。
connection.execute(select_statement) 执行查询。
result.fetchone() 获取结果集的第一行。
result.fetchall() 获取结果集的所有行（作为列表）。
result.scalar() 获取结果集第一行第一列的值。
迭代 result 对象会按行获取数据。

示例：更新数据 (Update)

“`python
from sqlalchemy import update

构造 update 语句

update_spongebob_name = update(users_table).where(users_table.c.name == “spongebob”).values(name=”Spongebob”)

with engine.connect() as connection:
result = connection.execute(update_spongebob_name)
connection.commit()
print(f”Rows matched for update: {result.rowcount}”) # rowcount 返回受影响的行数
“`

update(table) 创建一个更新表达式。
.where() 指定更新条件。
.values() 指定要更新的列和新值。

示例：删除数据 (Delete)

“`python
from sqlalchemy import delete

构造 delete 语句

delete_spongebob_addresses = delete(addresses_table).where(addresses_table.c.email_address.contains(“sqlalchemy”))

with engine.connect() as connection:
result = connection.execute(delete_spongebob_addresses)
connection.commit()
print(f”Rows deleted: {result.rowcount}”)

删除用户（如果开启了外键约束，需要先删除关联的地址或设置级联删除）

delete_spongebob_user = delete(users_table).where(users_table.c.name == “Spongebob”)

with engine.connect() as connection:

result = connection.execute(delete_spongebob_user)

connection.commit()

print(f”User rows deleted: {result.rowcount}”)

“`

delete(table) 创建一个删除表达式。
.where() 指定删除条件。

SQLAlchemy Core 提供了对 SQL 语言的强大抽象，它让你能够以更安全、更易读的方式构建复杂的 SQL 语句，并且不失 SQL 本身的灵活性。这对于需要高性能、细粒度控制数据库操作的场景非常有用。

SQLAlchemy ORM：对象关系映射器

ORM 是 SQLAlchemy 的高级部分，它允许我们将 Python 类映射到数据库表，将 Python 对象映射到数据库行。通过 ORM，我们可以用面向对象的方式来操作数据库，而 SQLAlchemy 会在背后完成对象和关系数据的转换。

ORM 的核心概念包括：

Declarative Mapping: 一种便捷的方式，通过继承一个基类并在类中定义属性来同时定义类和它到表的映射。
Mapped Classes / Models: 映射到数据库表的 Python 类。
Session (会话): ORM 的核心。它是一个与数据库进行对话的对象，管理着对象的加载、修改、删除、提交等操作，并且追踪对象的状态（工作单元模式）。
Relationships: 定义模型之间的关联，如一对多、多对多等。

1. Declarative Mapping：定义模型

使用 declarative_base() 创建一个基类，然后你的模型类继承自这个基类。在模型类中，使用 __tablename__ 指定对应的表名，使用 Column 和 ORM 特定的映射属性来定义列。

示例：使用 Declarative 定义模型

“`python
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship

创建声明性映射的基类

Base = declarative_base()

定义 User 模型，映射到 users 表

class User(Base):
tablename = “users” # 对应数据库的表名

id = Column(Integer, primary_key=True) # 主键
name = Column(String(50), nullable=False) # 非空字符串列
fullname = Column(String(100))

# 定义与 Address 模型的“一对多”关系
# 'addresses' 是在 User 对象上访问相关 Address 对象的属性名
# backref='user' 在 Address 对象上创建一个反向引用属性 'user'
addresses = relationship("Address", backref="user", cascade="all, delete-orphan")

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

定义 Address 模型，映射到 addresses 表

class Address(Base):
tablename = “addresses”

id = Column(Integer, primary_key=True)
email_address = Column(String(100), nullable=False)
# 定义外键列，关联 User 表的 id 列
user_id = Column(Integer, ForeignKey("users.id"), nullable=False)

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

print(“ORM Models defined.”)
“`

declarative_base() 创建了一个 Base 类，这个类包含了 MetaData 对象以及用于映射的类属性。
在模型类中，__tablename__ 是必需的，它告诉 ORM 这个类映射到哪个表。
类属性（如 id, name, fullname, user_id, email_address）使用 Column 定义，这与 Core 中定义列类似，但这里是作为类属性。
relationship() 是 ORM 的核心功能之一，用于定义模型之间的关系。它不是定义数据库中的外键约束（外键约束由 ForeignKey 在 Column 中定义），而是定义 Python 对象之间的关联。
- 在 User 类中定义 addresses = relationship("Address", ...) 意味着一个 User 对象可以通过 .addresses 属性访问一个 Address 对象的列表。
- backref='user' 参数会在关联的 Address 类上创建一个 .user 属性，通过这个属性可以方便地访问关联的 User 对象。
- cascade="all, delete-orphan" 是一个重要的级联操作设置。all 包括了 save-update (保存 User 时保存其关联的 Address) 和 delete (删除 User 时删除其关联的 Address)。delete-orphan 特别重要，它会在一个 Address 对象不再与任何 User 关联时（例如，将 Address 从 User 的 addresses 列表中移除）自动删除这个 Address 对象。

3. 创建表结构 (ORM)

与 Core 类似，ORM 模型也可以通过 Base.metadata.create_all() 创建数据库表。

示例：创建表 (ORM)

“`python

假设 engine 已经创建，且 Base 已经定义并包含了 User 和 Address 模型

Base.metadata.create_all(engine)

print(“ORM Tables created in the database.”)

“`

注意，Base.metadata 实际上就是 ORM 在内部使用的 MetaData 对象，它收集了所有继承自 Base 的模型所定义的表信息。

4. Session：与数据库交互的会话

ORM 的所有操作（增、删、改、查）都需要通过 Session 对象进行。Session 扮演着“工作单元”的角色，它加载对象、记录对象状态的变化，并在提交（commit）时将这些变化同步回数据库。

我们通常使用 sessionmaker 创建一个 Session 工厂，然后通过这个工厂创建 Session 实例。

示例：创建 Session 工厂和 Session

“`python
from sqlalchemy.orm import sessionmaker

创建 Session 工厂，绑定到 Engine

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

创建一个 Session 实例

db = SessionLocal()

在使用完毕后需要关闭 Session

db.close()

“`

sessionmaker 创建一个可调用的类，调用它可以创建 Session 实例。
bind=engine 将 Session 绑定到特定的 Engine。
autocommit=False (默认) 意味着需要手动调用 session.commit() 提交事务。
autoflush=False (默认 True) 控制何时将内存中的对象变更同步到数据库。通常保持默认 True 即可，在查询前会自动刷新变更。这里为了演示明确的提交/回滚，设置为 False。

使用 Session 的推荐方式（上下文管理器）

为了确保 Session 总是被正确关闭（即使发生错误），推荐使用上下文管理器：

“`python

from contextlib import contextmanager

@contextmanager

def get_db():

db = SessionLocal()

try:

yield db

finally:

db.close()

使用示例：

with get_db() as db:

# 在这里使用 db 对象进行 ORM 操作

pass

“`

在实际应用中，尤其是在 Web 框架中，Session 的管理通常会更规范化，比如每个请求一个 Session。

5. SQLAlchemy ORM CRUD 操作

通过 Session 对象，我们可以以面向对象的方式执行数据库操作。

示例：创建并添加对象

“`python

with get_db() as db: # 假设使用上下文管理器

# 创建对象实例

new_user = User(name=”patrick”, fullname=”Patrick Star”)

new_address = Address(email_address=”[email protected]”)

# 通过关系添加对象

new_user.addresses.append(new_address)

# 将对象添加到 Session

db.add(new_user) # add() 会追踪对象及其通过关系关联的对象 (new_address 也会被添加)

# 或者使用 add_all() 添加多个对象

# db.add_all([new_user, new_address]) # 如果没有通过关系关联，需要手动添加

# 提交事务，将对象持久化到数据库

db.commit()

# 提交后，对象的 id 会被填充

print(f”Created user: {new_user.id}”)

print(f”Created address: {new_address.id}”)

# 回滚事务 (如果发生错误或需要取消操作)

# db.rollback()

“`

创建模型类的实例 (User(...), Address(...))。
使用 session.add() 或 session.add_all() 将对象添加到 Session。对象此时处于“待定”（pending）状态。
使用 session.commit() 将 Session 中的变更（包括新添加的对象）同步到数据库并提交事务。成功提交后，新对象的自增主键会被填充。
使用 session.rollback() 取消 Session 中的所有未提交的变更。

示例：查询对象 (Select)

SQLAlchemy ORM 的查询接口经历了多次演变。早期使用 session.query()，而现代推荐使用基于 Core select() 表达式的 ORM 查询方式。这种新方式更灵活，并且与 Core 的表达方式一致。

使用基于 select() 的 ORM 查询 (推荐)

“`python
from sqlalchemy import select

假设已经创建了前面插入的用户和地址，并且 name=”Spongebob”, fullname=”Spongebob”

with get_db() as db:

# 查询所有用户

stmt = select(User) # select(Model) 查询该模型的所有列

users = db.execute(stmt).scalars().all() # execute() 执行，scalars() 获取模型对象，all() 获取所有结果

print(“\nAll users:”)

for user in users:

print(user)

# 按条件查询用户 (WHERE)

stmt = select(User).where(User.name == “Spongebob”)

spongebob = db.execute(stmt).scalar_one_or_none() # scalar_one_or_none() 期望返回 0 或 1 个结果，并返回其 scalar 值或 None

if spongebob:

print(f”\nFound Spongebob: {spongebob.fullname}”)

# 查询带有关联对象的用户 (JOIN)

# 使用 join() 方法，ORM 会自动处理连接条件（如果存在外键）

stmt = select(User).join(User.addresses).where(Address.email_address == “[email protected]”)

users_with_email = db.execute(stmt).scalars().all()

print(“\nUsers with specific email:”)

for user in users_with_email:

print(user)

# 查询特定列

stmt = select(User.name, User.fullname).where(User.name == “Spongebob”)

result = db.execute(stmt).first() # first() 获取第一行作为 ResultProxy

if result:

print(f”\nQuery specific columns: Name: {result.name}, Fullname: {result.fullname}”) # 通过列名访问

# 同时查询用户及其地址 (JOIN + 选择多个实体)

stmt = select(User, Address).join(User.addresses).where(User.name == “Spongebob”)

result = db.execute(stmt).all() # all() 获取所有行，每行是一个包含 User 和 Address 对象的元组

print(“\nUser and Address joined query:”)

for user, address in result:

print(f”User: {user.name}, Address: {address.email_address}”)

# 排序 (ORDER BY)

stmt = select(User).order_by(User.name)

users_ordered = db.execute(stmt).scalars().all()

print(“\nUsers ordered by name:”)

for user in users_ordered:

print(user.name)

# 限制结果数量 (LIMIT)

stmt = select(User).limit(1)

first_user = db.execute(stmt).scalar_one_or_none()

if first_user:

print(f”\nFirst user: {first_user.name}”)

# 偏移结果 (OFFSET)

stmt = select(User).offset(1).limit(1)

second_user = db.execute(stmt).scalar_one_or_none()

if second_user:

print(f”\nSecond user (offset 1): {second_user.name}”)

“`

select(Model) 创建查询模型对象的语句。
select(Model.column1, Model.column2) 查询特定列。
db.execute(stmt) 执行查询语句。
.scalars() 转换结果，使得每一行返回的是单个 ORM 对象（当 select 语句只选择了单个模型或单个 scalar 值时非常有用）。
.all() 获取所有结果。
.first() 获取第一行结果（作为 ResultProxy）。
.scalar_one_or_none() 期望结果只有 0 或 1 个 scalar 值。
.where(), .join(), .order_by(), .limit(), .offset() 等方法与 Core 中的用法类似，但在 ORM 中直接使用模型类和属性。

使用旧版 session.query() (了解即可，推荐使用 select())

“`python

# with get_db() as db:

# # 查询所有用户

# users = db.query(User).all()

# # 按条件查询

# spongebob = db.query(User).filter(User.name == “Spongebob”).first()

# # 查询带有关联对象

# users_with_email = db.query(User).join(Address).filter(Address.email_address == “[email protected]”).all()

# # 查询特定列

# result = db.query(User.name, User.fullname).filter(User.name == “Spongebob”).first()

“`

6. 更新对象

更新对象非常简单，只需要从 Session 中获取对象，修改其属性，然后提交 Session 即可。Session 会自动跟踪对象的修改。

示例：更新对象

“`python

with get_db() as db:

# 获取要更新的对象 (例如，根据 ID)

user_to_update = db.execute(select(User).where(User.name == “Spongebob”)).scalars().one() # one() 期望恰好返回 1 个结果

# 修改对象属性

user_to_update.fullname = “Spongebob Squarepants (Updated)”

# 添加一个新地址到关联列表 (如果 User 关系设置了 cascade=”all, delete-orphan”)

# new_address_for_spongebob = Address(email_address=”[email protected]”)

# user_to_update.addresses.append(new_address_for_spongebob)

# 提交 Session

db.commit()

print(f”User ‘{user_to_update.name}’ updated.”)

# 检查更新后的对象

# updated_user = db.execute(select(User).where(User.id == user_to_update.id)).scalars().one()

# print(f”Updated fullname: {updated_user.fullname}”)

# print(f”Updated addresses: {updated_user.addresses}”) # 会触发 addresses 的加载

“`

通过查询从 Session 中获取对象。
直接修改对象的属性。
调用 session.commit()。

7. 删除对象

从 Session 中删除对象使用 session.delete() 方法，然后提交 Session。

示例：删除对象

“`python

with get_db() as db:

# 获取要删除的对象 (例如，根据名称)

user_to_delete = db.execute(select(User).where(User.name == “Patrick Star”)).scalars().one_or_none()

if user_to_delete:

# 删除对象

db.delete(user_to_delete)

# 提交 Session

db.commit()

print(f”User ‘{user_to_delete.name}’ deleted.”)

else:

print(“User ‘Patrick Star’ not found.”)

“`

通过查询获取要删除的对象。
使用 session.delete(obj) 将对象标记为删除。
调用 session.commit() 执行删除操作。

8. 关系加载 (Loading Strategies)

当查询一个带有关系的对象时，SQLAlchemy ORM 如何加载相关的对象是一个重要的性能考虑。有几种加载策略：

Lazy Loading (懒加载): 这是默认策略。关联对象只有在第一次被访问时才会被加载。这可能会导致臭名昭著的 N+1 查询问题（例如，查询 N 个用户，然后遍历每个用户的地址，会导致 N+1 次数据库查询：1 次查询用户，N 次查询地址）。
Eager Loading (急加载): 在加载主对象时就立即加载关联对象。
- joinedload(): 使用 JOIN 查询一次性加载主对象和关联对象（适用于一对一、一对多，但不包含 many-to-many）。
- selectinload(): 对主对象执行一次查询，然后对所有关联对象执行一次额外的查询，使用 IN 子句加载。通常比 joinedload 更适合一对多关系，且对于多对多关系是常用的方式。
Subquery Load: 使用子查询加载。
NoLoad / RaiseLoad: 完全不加载或在访问时引发错误。

为了避免 N+1 问题，尤其是在需要访问关联对象的地方，应该使用急加载策略。

示例：使用急加载

“`python

with get_db() as db:

# 使用 joinedload 急加载 addresses 关系

from sqlalchemy.orm import joinedload

stmt_joined = select(User).options(joinedload(User.addresses))

users_with_addresses_joined = db.execute(stmt_joined).scalars().all()

print(“\nUsers loaded with joinedload (addresses):”)

for user in users_with_addresses_joined:

print(f”{user.name}’s addresses:”)

# 访问 user.addresses 不会触发新的查询，因为它们已经被加载

for address in user.addresses:

print(f”- {address.email_address}”)

# 使用 selectinload 急加载 addresses 关系 (通常更适合一对多)

from sqlalchemy.orm import selectinload

stmt_selectin = select(User).options(selectinload(User.addresses))

users_with_addresses_selectin = db.execute(stmt_selectin).scalars().all()

print(“\nUsers loaded with selectinload (addresses):”)

for user in users_with_addresses_selectin:

print(f”{user.name}’s addresses:”)

# 访问 user.addresses 也不会触发新的查询

for address in user.addresses:

print(f”- {address.email_address}”)

“`

理解并正确使用加载策略对于构建高性能的 ORM 应用至关重要。

SQLAlchemy Core vs. ORM：何时选择哪个？

SQLAlchemy Core 和 ORM 都有其适用的场景。它们不是互斥的，你可以在同一个应用中同时使用它们。

选择 SQLAlchemy Core:
- 需要对 SQL 有细粒度的控制: 当你需要编写复杂的、高度优化的 SQL 语句，或者使用数据库特有的功能时。
- 性能是首要考虑: Core 通常比 ORM 有更少的开销，因为它不涉及对象的映射和状态管理。对于大量数据的批量操作（插入、更新、删除），使用 Core 通常更高效。
- 处理存储过程、触发器等: Core 更接近底层的数据库操作。
- 应用程序非常简单，只需要基本的 CRUD 操作: 使用 Core 可能比引入 ORM 的开销更小。
- 处理元数据: Core 的 MetaData 对象是处理数据库模式信息的主要方式。
选择 SQLAlchemy ORM:
- 应用程序有复杂的数据模型和对象关系: ORM 极大地简化了处理对象之间的关系和导航。
- 倾向于面向对象编程: 你更喜欢用 Python 对象而不是 SQL 语句来思考和操作数据。
- 需要强大的抽象和自动化: ORM 处理了对象加载、状态跟踪、事务管理等许多繁琐的任务。
- 开发效率优先: 对于大多数业务应用，ORM 可以显著加快开发速度。
- 需要身份映射: ORM 确保在同一个 Session 中，同一个数据库行总是对应同一个 Python 对象实例。

在实践中，许多大型应用会同时使用 Core 和 ORM。ORM 用于处理大部分业务逻辑中的对象操作，而 Core 用于处理特定的高性能需求、批量操作或直接执行复杂的自定义 SQL。

最佳实践与进阶主题

Session 管理: 正确管理 Session 是 ORM 应用的关键。确保 Session 在请求处理结束时关闭或回滚，并避免在多个线程或请求之间共享同一个 Session 实例。使用上下文管理器是一个好的开始。
数据库迁移 (Alembic): 数据库模式会随着应用开发而演变。Alembic 是 SQLAlchemy 的官方数据库迁移工具，强烈推荐在 ORM 项目中使用它来管理数据库模式的变更。
连接池: SQLAlchemy 的 Engine 默认使用连接池，这对于应用程序性能非常重要。理解连接池的配置参数（如 pool_size, max_overflow, pool_recycle）并根据你的数据库和应用负载进行调整。
懒加载与急加载: 警惕 N+1 查询问题，并在需要时使用 joinedload, selectinload 等急加载策略。分析你的查询日志 (echo=True) 可以帮助发现 N+1 问题。
事务管理: 明确地使用 commit() 和 rollback() 管理事务。在使用上下文管理器 (with SessionLocal() as db:) 或 session.begin() 时，事务会自动处理（正常退出时提交，异常时回滚）。
数据类型映射: 了解 SQLAlchemy 通用类型如何映射到不同数据库的具体类型，以及如何定义自定义类型。
自定义 SQL / 原生查询: ORM 允许通过 session.execute() 执行 Core 表达式或原生 SQL 字符串，并将结果映射回 ORM 对象或行。这可以在 ORM 不足以表达特定需求时提供灵活性。

总结

SQLAlchemy 是一个功能极其丰富、高度灵活的 Python 数据库工具包。它提供了强大的 Core 抽象，允许开发者编写安全、高效的 SQL 表达式；同时也提供了功能完备的 ORM，让数据库操作变得面向对象且易于管理。

理解 SQLAlchemy 的 Core 和 ORM 两大部分及其核心概念（Engine, MetaData, Table, Column, Session, Model, Relationship）是掌握它的关键。通过合理地结合使用 Core 的低级控制和 ORM 的高级抽象，你可以构建出高效、可维护且强大的数据库驱动型 Python 应用程序。

虽然本文涵盖了 SQLAlchemy 的主要方面，但它还有更多高级功能和细节等待你去探索，例如事件系统、缓存、复杂查询技术、自定义类型等等。深入阅读官方文档将是你进一步提升 SQLAlchemy 技能的最佳途径。

希望这篇详细的文章能为你打开 SQLAlchemy 的大门，并帮助你更好地在 Python 项目中与数据库打交道。