SQLAlchemy 入门指南：Python ORM 基础与实践

SQLAlchemy 是 Python 中一个强大且流行的 SQL 工具包和对象关系映射器 (ORM)。它为应用程序开发者提供了操作数据库的灵活性，既可以进行底层的 SQL 操作，也可以通过 ORM 以面向对象的方式与数据库交互。

本文档主要关注 SQLAlchemy 的 ORM 部分，帮助新手快速上手。

1. 什么是 SQLAlchemy？

SQLAlchemy 提供了一套完整的企业级持久化模式，旨在实现高效和高性能的数据库访问。

1.1. SQLAlchemy Core

SQLAlchemy Core 是 SQLAlchemy 的基础部分。它提供了一套围绕 Pythonic SQL 表达式构建的工具。你可以使用它来直接编写和执行 SQL 语句，但它比直接使用 DBAPI (如 sqlite3, psycopg2) 更具抽象性和便利性。它不涉及对象映射。

1.2. SQLAlchemy ORM

SQLAlchemy ORM (Object Relational Mapper) 构建在 Core 之上。ORM 允许你将数据库中的表映射到 Python 中的类（称为模型 Model），将表中的行映射到这些类的实例（对象）。这样，你就可以通过操作 Python 对象来间接操作数据库，而不需要直接编写 SQL 语句（尽管也可以）。这是我们本文档的重点。

为什么使用 ORM？

开发效率：用面向对象的方式操作数据库，更符合 Python 的编程范式。
数据库无关性：理论上，更换数据库（如从 SQLite 换到 PostgreSQL）时，大部分 ORM 代码无需修改。
安全性：ORM 有助于防止 SQL 注入等安全问题。
代码可维护性：代码更易读、更易于组织。

2. 安装

首先，你需要安装 SQLAlchemy。推荐使用 pip：

bash

pip install sqlalchemy

如果你要连接特定的数据库（如 PostgreSQL, MySQL），你还需要安装相应的数据库驱动程序。例如：

PostgreSQL: pip install psycopg2-binary
MySQL: pip install mysqlclient or pip install PyMySQL
SQLite: Python 内置，无需额外安装驱动。

本文档将使用 SQLite 作为示例，因为它最简单，不需要额外配置。

3. 核心概念

在使用 SQLAlchemy ORM 之前，需要理解几个核心概念。

3.1. 引擎 (Engine)

引擎是 SQLAlchemy 应用中与数据库交互的起点。它负责处理数据库连接和执行 SQL 语句。引擎通常在应用程序启动时创建一次。

python

from sqlalchemy import create_engine

# SQLite 内存数据库 (程序结束数据丢失)
# engine = create_engine("sqlite:///:memory:")

# SQLite 文件数据库 (数据持久化到文件)
# 语法: "dialect+driver://username:password@host:port/database"
engine = create_engine("sqlite:///mydatabase.db") # mydatabase.db 将在当前目录创建

sqlite:///mydatabase.db：这是数据库连接字符串 (Connection String)。
- sqlite: 表示使用的是 SQLite 数据库。
- ///mydatabase.db: 表示数据库文件名为 mydatabase.db，位于当前路径。如果是绝对路径，会是 sqlite:////path/to/mydatabase.db (注意四个斜杠)。

3.2. 声明式基类 (Declarative Base)

在 ORM 中，我们需要定义 Python 类来映射数据库中的表。declarative_base 是一个工厂函数，它返回一个基类，我们定义的模型类将继承这个基类。

python

from sqlalchemy.orm import declarative_base

Base = declarative_base()

所有的数据模型（表）都将继承自这个 Base。

3.3. 模型 (Model)

模型是代表数据库中一个表的 Python 类。它继承自上面创建的 Base。模型的属性 (attributes) 通过 Column 对象映射到表的列 (columns)。

python

from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.orm import declarative_base
import datetime

Base = declarative_base()

class User(Base):
    __tablename__ = "users"  # 数据库中对应的表名

    # 定义列
    id = Column(Integer, primary_key=True, index=True) # 主键，自动索引
    name = Column(String(50), nullable=False) # 字符串类型，长度50，不能为空
    email = Column(String(100), unique=True, index=True) # 字符串类型，长度100，唯一，索引
    age = Column(Integer, nullable=True)
    created_at = Column(DateTime, default=datetime.datetime.utcnow) # 创建时间，默认为当前UTC时间

    # 可选: 定义一个易于阅读的字符串表示形式，方便调试
    def __repr__(self):
        return f"<User(id={self.id}, name='{self.name}', email='{self.email}')>"

__tablename__: 指定这个模型映射到数据库中的表名。
Column: 定义表的列。
- Integer, String, DateTime 等是 SQLAlchemy 提供的列类型。
- primary_key=True: 指定该列为主键。
- index=True: 为该列创建索引，加快查询速度。
- nullable=False: 指定该列不能为空。
- unique=True: 指定该列的值必须唯一。
- default: 指定列的默认值。

3.4. 会话 (Session)

会话是 ORM 与数据库进行交互的主要接口。你可以把它想象成一个“工作区”或“暂存区”。所有对数据库的更改（添加、修改、删除对象）都是先在会话中进行的，然后通过会话的 commit() 方法一次性写入数据库。

会话的创建通常通过 sessionmaker 完成，它是一个会话工厂。

python

from sqlalchemy.orm import sessionmaker

# 创建 SessionLocal 类，它将用于创建实际的会话实例
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

autocommit=False: 自动提交关闭。我们需要手动调用 session.commit()。
autoflush=False: 自动刷新关闭。flush 是指将会话中的更改发送到数据库，但尚未提交事务。
bind=engine: 将这个会话工厂绑定到我们之前创建的数据库引擎。

4. 基础操作：CURD

CURD 代表创建 (Create)、读取 (Read)、更新 (Update) 和删除 (Delete)，这是数据库操作的四个基本功能。

4.1. 连接数据库与创建表

在进行任何 CURD 操作之前，我们需要确保数据库表已经根据我们的模型定义创建好了。

python

from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.orm import declarative_base, sessionmaker
import datetime

# --- 1. 定义引擎 ---
engine = create_engine("sqlite:///mydatabase.db", echo=True) # echo=True 会打印执行的SQL语句，方便调试

# --- 2. 定义声明式基类 ---
Base = declarative_base()

# --- 3. 定义模型 ---
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50), nullable=False)
    email = Column(String(100), unique=True, index=True)
    age = Column(Integer, nullable=True)
    created_at = Column(DateTime, default=datetime.datetime.utcnow)

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

# --- 4. 创建表 (如果表已存在，则不会重复创建) ---
# 这应该在应用程序初始化时执行一次
Base.metadata.create_all(bind=engine)
print("表已创建 (如果不存在的话)")

# --- 5. 创建会话工厂 ---
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# --- 辅助函数：获取一个数据库会话 ---
def get_db_session():
    db = SessionLocal()
    try:
        yield db  # 使用 yield 使其成为一个生成器，方便与 with 语句配合
    finally:
        db.close() # 确保会话总是被关闭

Base.metadata.create_all(bind=engine) 会检查数据库中是否存在名为 users 的表，如果不存在，则根据 User 模型的定义创建它。如果表已存在，并且结构与模型定义一致，则不会做任何事情。

4.2. C - 创建 (Create) / 添加数据

要向数据库中添加一条新记录，你需要：

创建一个模型类的实例。
将这个实例添加到会话中 (session.add())。
提交会话 (session.commit()) 将更改保存到数据库。

python

# 获取一个会话
db_session_gen = get_db_session()
db = next(db_session_gen) # 从生成器获取会话实例

try:
    # 创建 User 对象
    new_user_1 = User(name="Alice Wonderland", email="alice@example.com", age=30)
    new_user_2 = User(name="Bob The Builder", email="bob@example.com", age=45)

    # 将对象添加到会话
    db.add(new_user_1)
    db.add(new_user_2)
    # 也可以使用 db.add_all([new_user_1, new_user_2]) 添加多个

    # 提交事务到数据库
    db.commit()
    print(f"添加成功: {new_user_1}")
    print(f"添加成功: {new_user_2}")

    # 提交后，对象会被赋予 ID (如果是自增主键)
    print(f"Alice's ID: {new_user_1.id}")
    print(f"Bob's ID: {new_user_2.id}")

except Exception as e:
    db.rollback() # 如果发生错误，回滚事务
    print(f"添加失败: {e}")
finally:
    db.close() # 关闭会话

注意：

db.add(object)：将对象标记为“待添加”。
db.commit()：实际将所有标记的更改（添加、更新、删除）写入数据库。如果发生错误，它会回滚所有未提交的更改。
db.rollback()：如果 commit() 过程中发生错误，或者你想主动取消更改，可以调用此方法。

4.3. R - 读取 (Read) / 查询数据

查询数据是数据库操作中最常用的功能。SQLAlchemy 提供了强大的查询 API。查询通常以 session.query(Model) 开始。

4.3.1. 查询所有记录

使用 .all() 方法获取查询结果的所有记录，返回一个列表。

python

db = next(get_db_session())
try:
    print("\n--- 查询所有用户 ---")
    all_users = db.query(User).all()
    if not all_users:
        print("数据库中没有用户。")
    for user in all_users:
        print(user)
finally:
    db.close()

4.3.2. 查询第一条记录

使用 .first() 方法获取查询结果的第一条记录，如果查询没有结果，则返回 None。

python

db = next(get_db_session())
try:
    print("\n--- 查询第一个用户 ---")
    first_user = db.query(User).first()
    if first_user:
        print(first_user)
    else:
        print("数据库中没有用户。")
finally:
    db.close()

4.3.3. 根据主键查询

使用 .get(primary_key_value) 方法，这是通过主键获取单个对象的最快方式。

python

db = next(get_db_session())
try:
    print("\n--- 根据 ID 查询用户 (假设 ID=1 存在) ---")
    user_by_id = db.query(User).get(1) # 假设 Alice 的 ID 是 1
    if user_by_id:
        print(user_by_id)
    else:
        print("未找到 ID 为 1 的用户。")
finally:
    db.close()

4.3.4. 条件查询 (Filtering)

使用 .filter() 或 .filter_by() 方法添加查询条件。

.filter_by(attribute=value): 简单等值查询。
.filter(Model.attribute operator value): 更灵活，可以使用各种比较运算符。

python

db = next(get_db_session())
try:
    print("\n--- 条件查询 ---")
    # 1. 使用 filter_by (简单等值)
    alice = db.query(User).filter_by(name="Alice Wonderland").first()
    if alice:
        print(f"通过 filter_by 找到 Alice: {alice}")

    # 2. 使用 filter (更灵活)
    #    查询年龄大于 30 的用户
    users_older_than_30 = db.query(User).filter(User.age > 30).all()
    print("年龄大于 30 的用户:")
    for user in users_older_than_30:
        print(user)

    #    查询 email 包含 'example.com' 的用户 (模糊查询)
    users_with_example_email = db.query(User).filter(User.email.like("%example.com%")).all()
    print("Email 包含 'example.com' 的用户:")
    for user in users_with_example_email:
        print(user)

    #    组合条件 (AND): 链式调用 filter 或 filter_by
    user_bob_older_40 = db.query(User).filter(User.name == "Bob The Builder").filter(User.age > 40).first()
    if user_bob_older_40:
        print(f"名叫 Bob 且年龄大于 40: {user_bob_older_40}")

    #    组合条件 (OR): 使用 or_
    from sqlalchemy import or_
    users_alice_or_older_40 = db.query(User).filter(or_(User.name == "Alice Wonderland", User.age > 40)).all()
    print("名叫 Alice 或年龄大于 40 的用户:")
    for user in users_alice_or_older_40:
        print(user)

finally:
    db.close()

常用操作符：

==: 等于 (如 User.name == "Alice")
!=: 不等于 (如 User.age != 30)
<, >, <=, >=: 小于，大于，小于等于，大于等于
.like("pattern"): SQL LIKE 操作，% 是通配符 (如 User.name.like("A%") 查找以 A 开头的名字)
.in_([val1, val2]): 属于列表中的值 (如 User.name.in_(["Alice", "Bob"]))
.is_(None) 或 .isnot(None): 判断是否为 NULL (如 User.age.is_(None))

4.3.5. 排序 (Ordering)

使用 .order_by() 方法对结果进行排序。

Model.attribute.asc(): 升序。
Model.attribute.desc(): 降序。

python

db = next(get_db_session())
try:
    print("\n--- 排序查询 ---")
    # 按年龄降序排列
    users_by_age_desc = db.query(User).order_by(User.age.desc()).all()
    print("按年龄降序:")
    for user in users_by_age_desc:
        print(user)

    # 按创建时间升序排列
    users_by_created_asc = db.query(User).order_by(User.created_at.asc()).all()
    print("按创建时间升序:")
    for user in users_by_created_asc:
        print(user)
finally:
    db.close()

4.3.6. 限制数量 (Limiting) 与偏移 (Offsetting)

.limit(n): 限制返回结果的数量为 n 条。
.offset(m): 跳过前 m 条结果。

这常用于分页查询。

python

db = next(get_db_session())
try:
    print("\n--- 分页查询 (每页1条，查询第2页) ---")
    # 假设我们每页显示1条记录，我们想看第2页
    page_size = 1
    page_number = 2
    users_page_2 = db.query(User).order_by(User.id.asc()).limit(page_size).offset((page_number - 1) * page_size).all()
    print(f"第 {page_number} 页的用户:")
    for user in users_page_2:
        print(user)
finally:
    db.close()

4.4. U - 更新 (Update) / 修改数据

要更新一条记录：

首先，查询到要更新的对象。
直接修改对象的属性。
提交会话 (session.commit())。SQLAlchemy 会自动检测到对象的更改。

python

db = next(get_db_session())
try:
    print("\n--- 更新用户数据 ---")
    # 假设我们要更新 Alice 的年龄
    user_to_update = db.query(User).filter_by(name="Alice Wonderland").first()

    if user_to_update:
        print(f"更新前: {user_to_update}, 年龄: {user_to_update.age}")
        user_to_update.age = 31 # 修改属性
        user_to_update.email = "alice.w@newdomain.com" # 修改属性
        db.commit() # 提交更改
        print(f"更新后: {user_to_update}, 年龄: {user_to_update.age}, Email: {user_to_update.email}")

        # 验证一下，重新从数据库查询
        updated_user_from_db = db.query(User).get(user_to_update.id)
        print(f"从数据库重新查询: {updated_user_from_db}, 年龄: {updated_user_from_db.age}")
    else:
        print("未找到 Alice Wonderland 进行更新。")

except Exception as e:
    db.rollback()
    print(f"更新失败: {e}")
finally:
    db.close()

批量更新：如果你想不加载对象就更新数据（例如，将所有年龄为空的用户年龄设置为18），可以使用 query.update()：

python

db = next(get_db_session())
try:
    print("\n--- 批量更新 ---")
    # 将所有 age 为 NULL 的用户的 age 更新为 18
    # {User.age: 18} 表示将 User.age 列更新为 18
    # synchronize_session=False 表示不要将会话中的对象与更新同步，
    # 对于简单更新，通常设为 'evaluate' 或 False。
    # 'evaluate': SQLAlchemy会尝试智能地更新内存中的对象。
    # 'fetch': 会先查询匹配的对象，然后更新。
    # False: 不更新内存中的对象，性能较高，但需注意会话中对象可能与数据库不一致。
    updated_count = db.query(User).filter(User.age.is_(None)).update({User.age: 18}, synchronize_session=False)
    db.commit()
    print(f"批量更新了 {updated_count} 条记录的年龄为 18。")

    # 验证
    users_with_age_18 = db.query(User).filter_by(age=18).all()
    print("现在年龄为 18 的用户:")
    for user in users_with_age_18:
        print(user)
except Exception as e:
    db.rollback()
    print(f"批量更新失败: {e}")
finally:
    db.close()

4.5. D - 删除 (Delete) / 删除数据

要删除一条记录：

查询到要删除的对象。
使用 session.delete(object) 将其标记为“待删除”。
提交会话 (session.commit())。

python

db = next(get_db_session())
try:
    print("\n--- 删除用户数据 ---")
    # 假设我们要删除 Bob
    user_to_delete = db.query(User).filter_by(name="Bob The Builder").first()

    if user_to_delete:
        print(f"准备删除: {user_to_delete}")
        db.delete(user_to_delete) # 标记为待删除
        db.commit() # 提交更改
        print(f"用户 '{user_to_delete.name}' 已删除。")

        # 验证一下，尝试再次查询
        deleted_user_check = db.query(User).filter_by(name="Bob The Builder").first()
        if not deleted_user_check:
            print("确认 Bob 已被删除。")
    else:
        print("未找到 Bob The Builder 进行删除。")
except Exception as e:
    db.rollback()
    print(f"删除失败: {e}")
finally:
    db.close()

批量删除：与批量更新类似，可以使用 query.delete()：

python

db = next(get_db_session())
try:
    print("\n--- 批量删除 ---")
    # 删除所有年龄大于 50 的用户
    # synchronize_session=False 同样适用
    deleted_count = db.query(User).filter(User.age > 50).delete(synchronize_session=False)
    db.commit()
    print(f"批量删除了 {deleted_count} 条年龄大于 50 的记录。")

    # 验证
    remaining_users = db.query(User).all()
    print("剩余用户:")
    for user in remaining_users:
        print(user)
except Exception as e:
    db.rollback()
    print(f"批量删除失败: {e}")
finally:
    db.close()

5. 会话管理进阶

5.1. 提交 (Commit) 与回滚 (Rollback)

session.commit():
- 将当前会话中所有挂起的更改（添加、更新、删除）永久保存到数据库。
- 在一个事务的上下文中执行。如果成功，事务被提交；如果失败，事务被回滚。
- 提交后，会话中的对象状态会更新（例如，新对象的 ID 会被填充）。
- 它也会结束当前的事务，并开启一个新的事务（如果会话继续使用）。
session.rollback():
- 撤销当前事务中所有未提交的更改。
- 将会话中的对象恢复到事务开始前的状态。
- 通常在发生异常时调用，以确保数据的一致性。

5.2. 使用 `with` 语句管理会话

之前我们使用了 try...finally...db.close() 的模式来确保会话被关闭。一个更 Pythonic 的方式是使用 with 语句，这需要我们的会话工厂或获取会话的函数支持上下文管理协议。

我们可以修改 get_db_session 函数，或者直接使用 SessionLocal 作为上下文管理器（如果它被设计成这样，但 sessionmaker 返回的类默认不是）。更常见的是创建一个上下文管理的函数：

python

from contextlib import contextmanager

@contextmanager
def get_db(): # 与之前的 get_db_session 作用类似，但更适合 with
    db = SessionLocal()
    try:
        yield db
        db.commit() # 如果 with 块成功执行，则提交
    except Exception:
        db.rollback() # 如果 with 块中发生异常，则回滚
        raise # 重新抛出异常，以便上层代码知道出错了
    finally:
        db.close() # 无论如何都关闭会话

现在可以这样使用：

python

# 使用 with 语句进行创建操作
try:
    with get_db() as db: # db 就是会话实例
        new_user_charlie = User(name="Charlie Brown", email="charlie@example.com", age=8)
        db.add(new_user_charlie)
        # commit 会在 with 块成功结束时自动调用
    print(f"通过 with 添加 Charlie 成功: {new_user_charlie.id}")
except Exception as e:
    print(f"使用 with 添加 Charlie 失败: {e}")


# 使用 with 语句进行查询操作
try:
    with get_db() as db:
        charlie = db.query(User).filter_by(name="Charlie Brown").first()
        if charlie:
            print(f"查询到 Charlie: {charlie}")
        else:
            print("未找到 Charlie")
        # 对于只读操作，上面的 get_db 实现也会 commit，这无害但非必需。
        # 如果是纯读取，可以考虑一个不自动 commit 的版本。
except Exception as e:
    print(f"使用 with 查询 Charlie 失败: {e}")

这种 with 结构使得代码更简洁，并且自动处理了提交、回滚和关闭会话的逻辑。

6. 总结

本文档介绍了 SQLAlchemy ORM 的基础知识，包括：

核心概念：Engine, Declarative Base, Model, Session。
表的创建：Base.metadata.create_all()。
CURD 操作：
- Create: session.add(), session.commit()
- Read: session.query().all(), .first(), .get(), .filter(), .filter_by(), .order_by(), .limit(), .offset()
- Update: 查询对象 -> 修改属性 -> session.commit() 或 query.update()
- Delete: 查询对象 -> session.delete() -> session.commit() 或 query.delete()
会话管理：commit(), rollback(), 以及使用 with 语句进行更优雅的会话管理。

SQLAlchemy 是一个功能非常丰富的库，本文仅触及皮毛。随着你对它的深入，还可以学习到：

复杂关系映射 (一对多、多对多等)。
更高级的查询技巧。
数据库迁移 (配合 Alembic)。
异步 SQLAlchemy (用于 ASGI 应用如 FastAPI)。

希望这份文档能为你学习 SQLAlchemy 打下坚实的基础！多动手实践是掌握它的最好方法。

SQLAlchemy 入门指南：Python ORM 基础与实践 ​

目录 ​

1. 什么是 SQLAlchemy？ ​

1.1. SQLAlchemy Core ​

1.2. SQLAlchemy ORM ​

2. 安装 ​

3. 核心概念 ​

3.1. 引擎 (Engine) ​

3.2. 声明式基类 (Declarative Base) ​

3.3. 模型 (Model) ​

3.4. 会话 (Session) ​

4. 基础操作：CURD ​

4.1. 连接数据库与创建表 ​

4.2. C - 创建 (Create) / 添加数据 ​

4.3. R - 读取 (Read) / 查询数据 ​

4.3.1. 查询所有记录 ​

4.3.2. 查询第一条记录 ​

4.3.3. 根据主键查询 ​

4.3.4. 条件查询 (Filtering) ​

4.3.5. 排序 (Ordering) ​

4.3.6. 限制数量 (Limiting) 与偏移 (Offsetting) ​

4.4. U - 更新 (Update) / 修改数据 ​

4.5. D - 删除 (Delete) / 删除数据 ​

5. 会话管理进阶 ​

5.1. 提交 (Commit) 与回滚 (Rollback) ​

5.2. 使用 with 语句管理会话 ​

6. 总结 ​

SQLAlchemy 入门指南：Python ORM 基础与实践

目录

1. 什么是 SQLAlchemy？

1.1. SQLAlchemy Core

1.2. SQLAlchemy ORM

2. 安装

3. 核心概念

3.1. 引擎 (Engine)

3.2. 声明式基类 (Declarative Base)

3.3. 模型 (Model)

3.4. 会话 (Session)

4. 基础操作：CURD

4.1. 连接数据库与创建表

4.2. C - 创建 (Create) / 添加数据

4.3. R - 读取 (Read) / 查询数据

4.3.1. 查询所有记录

4.3.2. 查询第一条记录

4.3.3. 根据主键查询

4.3.4. 条件查询 (Filtering)

4.3.5. 排序 (Ordering)

4.3.6. 限制数量 (Limiting) 与偏移 (Offsetting)

4.4. U - 更新 (Update) / 修改数据

4.5. D - 删除 (Delete) / 删除数据

5. 会话管理进阶

5.1. 提交 (Commit) 与回滚 (Rollback)

5.2. 使用 `with` 语句管理会话

6. 总结