对于希望快速了解基本 ORM 使用方法的新用户，这里简要介绍了 SQLAlchemy 统一教程中使用的映射和示例。此处的代码可以在干净的命令行中完全运行。

由于本节中的描述故意很简短，请参阅完整的 SQLAlchemy 统一教程，以更深入地了解此处说明的每个概念。

声明模型¶

在这里，我们定义模块级构造，这些构造将构成我们从数据库中查询的结构。这种结构被称为声明式映射，它同时定义了 Python 对象模型以及数据库元数据，该元数据描述了在特定数据库中存在或将存在的实际 SQL 表。

>>> from typing import List
>>> from typing import Optional
>>> from sqlalchemy import ForeignKey
>>> from sqlalchemy import String
>>> from sqlalchemy.orm import DeclarativeBase
>>> from sqlalchemy.orm import Mapped
>>> from sqlalchemy.orm import mapped_column
>>> from sqlalchemy.orm import relationship

>>> class Base(DeclarativeBase):
...     pass

>>> class User(Base):
...     __tablename__ = "user_account"
...
...     id: Mapped[int] = mapped_column(primary_key=True)
...     name: Mapped[str] = mapped_column(String(30))
...     fullname: Mapped[Optional[str]]
...
...     addresses: Mapped[List["Address"]] = relationship(
...         back_populates="user", cascade="all, delete-orphan"
...     )
...
...     def __repr__(self) -> str:
...         return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"

>>> class Address(Base):
...     __tablename__ = "address"
...
...     id: Mapped[int] = mapped_column(primary_key=True)
...     email_address: Mapped[str]
...     user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
...
...     user: Mapped["User"] = relationship(back_populates="addresses")
...
...     def __repr__(self) -> str:
...         return f"Address(id={self.id!r}, email_address={self.email_address!r})"

映射从一个基类开始，上面称为 Base，它是通过对 DeclarativeBase 类进行简单子类化创建的。

然后，通过对 Base 进行子类化来创建各个映射类。映射类通常对应于单个特定的数据库表，该表的名称由使用 __tablename__ 类级属性来指示。

接下来，声明作为表一部分的列，方法是添加包含名为 Mapped 的特殊类型注释的属性。每个属性的名称对应于将成为数据库表一部分的列。每列的数据类型首先取自与每个 Mapped 注释关联的 Python 数据类型；int 表示 INTEGER，str 表示 VARCHAR，等等。可空性取决于是否使用 Optional[] 类型修饰符。可以使用 SQLAlchemy 类型对象在右侧 mapped_column() 指令中指示更具体的类型信息，例如上面在 User.name 列中使用的 String 数据类型。Python 类型和 SQL 类型之间的关联可以使用类型注释映射进行自定义。

所有需要更具体自定义的基于列的属性都使用 mapped_column() 指令。除了类型信息外，此指令还接受各种参数，这些参数指示有关数据库列的具体细节，包括服务器默认值和约束信息，例如作为主键和外键的成员资格。 mapped_column() 指令接受 SQLAlchemy Column 类接受的参数的超集，该类由 SQLAlchemy Core 用于表示数据库列。

所有 ORM 映射类都需要至少声明一列作为主键的一部分，通常使用 Column.primary_key 参数在那些应作为主键一部分的 mapped_column() 对象上。在上面的示例中，User.id 和 Address.id 列被标记为主键。

字符串表名与列声明列表的组合在 SQLAlchemy 中被称为表元数据。在 SQLAlchemy 统一教程的使用数据库元数据中介绍了使用 Core 和 ORM 方法设置表元数据。上面的映射是带注释的声明式表配置的示例。

还有其他类型的 Mapped 可用，最常见的是上面所示的 relationship() 构造。与基于列的属性不同，relationship() 表示两个 ORM 类之间的链接。在上面的示例中，User.addresses 将 User 链接到 Address，而 Address.user 将 Address 链接到 User。 relationship() 构造在 SQLAlchemy 统一教程的使用 ORM 相关对象中进行了介绍。

最后，上面的示例类包括一个 __repr__() 方法，该方法不是必需的，但对于调试很有用。可以使用诸如 __repr__() 之类的自动生成的方法来创建映射类，使用 dataclasses。有关 dataclass 映射的更多信息，请参阅声明式 Dataclass 映射。

创建引擎¶

Engine 是一个工厂，可以为我们创建新的数据库连接，它还将连接保存在连接池中，以便快速重复使用。为了学习目的，我们通常使用 SQLite 内存数据库，因为它很方便

>>> from sqlalchemy import create_engine
>>> engine = create_engine("sqlite://", echo=True)

提示

echo=True 参数表示连接发出的 SQL 将被记录到标准输出中。

有关 Engine 的完整介绍，请从建立连接 - 引擎开始。

发出 CREATE TABLE DDL¶

使用我们的表元数据和引擎，我们可以使用名为 MetaData.create_all() 的方法立即在我们的目标 SQLite 数据库中生成我们的模式

>>> Base.metadata.create_all(engine)
BEGIN (implicit)
PRAGMA main.table_...info("user_account")
...
PRAGMA main.table_...info("address")
...
CREATE TABLE user_account (
    id INTEGER NOT NULL,
    name VARCHAR(30) NOT NULL,
    fullname VARCHAR,
    PRIMARY KEY (id)
)
...
CREATE TABLE address (
    id INTEGER NOT NULL,
    email_address VARCHAR NOT NULL,
    user_id INTEGER NOT NULL,
    PRIMARY KEY (id),
    FOREIGN KEY(user_id) REFERENCES user_account (id)
)
...
COMMIT

我们刚刚写的那段 Python 代码做了很多事情。要全面了解表元数据的工作原理，请在教程的使用数据库元数据中继续学习。

创建对象并持久化¶

现在我们准备将数据插入数据库。我们通过创建 User 和 Address 类的实例来实现这一点，这些类已经具有一个 __init__() 方法，该方法由声明式映射过程自动建立。然后，我们使用名为 Session 的对象将它们传递到数据库，该对象利用 Engine 与数据库交互。 Session.add_all() 方法用于一次添加多个对象，而 Session.commit() 方法将用于冲刷对数据库的所有待处理更改，然后提交当前数据库事务，该事务在使用 Session 时始终处于进行中。

>>> from sqlalchemy.orm import Session

>>> with Session(engine) as session:
...     spongebob = User(
...         name="spongebob",
...         fullname="Spongebob Squarepants",
...         addresses=[Address(email_address="spongebob@sqlalchemy.org")],
...     )
...     sandy = User(
...         name="sandy",
...         fullname="Sandy Cheeks",
...         addresses=[
...             Address(email_address="sandy@sqlalchemy.org"),
...             Address(email_address="sandy@squirrelpower.org"),
...         ],
...     )
...     patrick = User(name="patrick", fullname="Patrick Star")
...
...     session.add_all([spongebob, sandy, patrick])
...
...     session.commit()
BEGIN (implicit)
INSERT INTO user_account (name, fullname) VALUES (?, ?) RETURNING id
[...] ('spongebob', 'Spongebob Squarepants')
INSERT INTO user_account (name, fullname) VALUES (?, ?) RETURNING id
[...] ('sandy', 'Sandy Cheeks')
INSERT INTO user_account (name, fullname) VALUES (?, ?) RETURNING id
[...] ('patrick', 'Patrick Star')
INSERT INTO address (email_address, user_id) VALUES (?, ?) RETURNING id
[...] ('spongebob@sqlalchemy.org', 1)
INSERT INTO address (email_address, user_id) VALUES (?, ?) RETURNING id
[...] ('sandy@sqlalchemy.org', 2)
INSERT INTO address (email_address, user_id) VALUES (?, ?) RETURNING id
[...] ('sandy@squirrelpower.org', 2)
COMMIT

提示

建议使用上面所示的上下文管理器风格使用 Session，即使用 Python with: 语句。 Session 对象表示活动数据库资源，因此最好确保在完成一系列操作后关闭它。在下一节中，我们将仅出于说明目的打开一个 Session。

有关创建 Session 的基础知识，请参阅使用 ORM 会话执行，以及使用会话的基础知识。

然后，在使用 ORM 工作单元模式插入行中介绍了各种基本的持久化操作。

简单 SELECT¶

在数据库中存在一些行之后，这是发出 SELECT 语句以加载一些对象的简单形式。要创建 SELECT 语句，我们使用 select() 函数创建一个新的 Select 对象，然后我们使用 Session 调用它。当查询 ORM 对象时，通常很有用的方法是 Session.scalars() 方法，它将返回一个 ScalarResult 对象，该对象将遍历我们选择的 ORM 对象。

>>> from sqlalchemy import select

>>> session = Session(engine)

>>> stmt = select(User).where(User.name.in_(["spongebob", "sandy"]))

>>> for user in session.scalars(stmt):
...     print(user)
BEGIN (implicit)
SELECT user_account.id, user_account.name, user_account.fullname
FROM user_account
WHERE user_account.name IN (?, ?)
[...] ('spongebob', 'sandy')
User(id=1, name='spongebob', fullname='Spongebob Squarepants')
User(id=2, name='sandy', fullname='Sandy Cheeks')

上面的查询还使用了 Select.where() 方法添加 WHERE 条件，还使用了 ColumnOperators.in_() 方法，该方法是所有 SQLAlchemy 类列构造的一部分，用于使用 SQL IN 运算符。

有关如何选择对象和单个列的更多详细信息，请参阅选择 ORM 实体和列。

使用 JOIN 的 SELECT¶

在多个表中同时查询非常常见，在 SQL 中，JOIN 关键字是实现此目的的主要方法。 Select 构造使用 Select.join() 方法创建连接。

>>> stmt = (
...     select(Address)
...     .join(Address.user)
...     .where(User.name == "sandy")
...     .where(Address.email_address == "sandy@sqlalchemy.org")
... )
>>> sandy_address = session.scalars(stmt).one()
SELECT address.id, address.email_address, address.user_id
FROM address JOIN user_account ON user_account.id = address.user_id
WHERE user_account.name = ? AND address.email_address = ?
[...] ('sandy', 'sandy@sqlalchemy.org')
>>> sandy_address
Address(id=2, email_address='sandy@sqlalchemy.org')

上面的查询说明了多个 WHERE 条件，这些条件使用 AND 自动链接在一起，以及如何使用 SQLAlchemy 类列对象创建“相等”比较，这些对象使用重写的 Python 方法 ColumnOperators.__eq__() 生成 SQL 条件对象。

有关上述概念的一些更多背景信息，请参阅 WHERE 子句和显式 FROM 子句和 JOIN。

进行更改¶

Session 对象与我们的 ORM 映射类 User 和 Address 一同自动跟踪对对象的更改，这些更改会导致在 Session 下次刷新时发出的 SQL 语句。在下文中，我们更改了与“sandy”关联的一个电子邮件地址，并在发出 SELECT 以检索“patrick”的行后，还向“patrick”添加了一个新的电子邮件地址。

>>> stmt = select(User).where(User.name == "patrick")
>>> patrick = session.scalars(stmt).one()
SELECT user_account.id, user_account.name, user_account.fullname
FROM user_account
WHERE user_account.name = ?
[...] ('patrick',)
>>> patrick.addresses.append(Address(email_address="patrickstar@sqlalchemy.org"))
SELECT address.id AS address_id, address.email_address AS address_email_address, address.user_id AS address_user_id
FROM address
WHERE ? = address.user_id
[...] (3,)
>>> sandy_address.email_address = "sandy_cheeks@sqlalchemy.org"

>>> session.commit()
UPDATE address SET email_address=? WHERE address.id = ?
[...] ('sandy_cheeks@sqlalchemy.org', 2)
INSERT INTO address (email_address, user_id) VALUES (?, ?)
[...] ('patrickstar@sqlalchemy.org', 3)
COMMIT

请注意，当我们访问 patrick.addresses 时，发出了一个 SELECT。这称为延迟加载。有关使用更多或更少 SQL 访问相关项目的不同方法的背景信息，请参阅加载策略。

有关 ORM 数据操作的详细介绍，请参阅使用 ORM 进行数据操作。

一些删除操作¶

凡事皆有终点，我们的数据库行也是如此——这里简要演示了两种不同的删除形式，这两种形式根据具体的用例都很重要。

首先，我们将从“sandy”用户中删除一个 Address 对象。当 Session 下次刷新时，这将导致该行被删除。此行为是我们映射中配置的一项功能，称为删除级联。我们可以使用 Session.get() 通过主键获取 sandy 对象的句柄，然后处理该对象。

>>> sandy = session.get(User, 2)
BEGIN (implicit)
SELECT user_account.id AS user_account_id, user_account.name AS user_account_name, user_account.fullname AS user_account_fullname
FROM user_account
WHERE user_account.id = ?
[...] (2,)
>>> sandy.addresses.remove(sandy_address)
SELECT address.id AS address_id, address.email_address AS address_email_address, address.user_id AS address_user_id
FROM address
WHERE ? = address.user_id
[...] (2,)

上面的最后一个 SELECT 是延迟加载操作的进行，以便可以加载 sandy.addresses 集合，以便我们可以删除 sandy_address 成员。还有其他方法可以执行这一系列操作，这些方法不会发出那么多的 SQL。

我们可以选择发出将要更改的内容的 DELETE SQL，而不提交事务，可以使用 Session.flush() 方法。

>>> session.flush()
DELETE FROM address WHERE address.id = ?
[...] (2,)

接下来，我们将完全删除“patrick”用户。对于对对象本身进行的顶层删除，我们使用 Session.delete() 方法；该方法实际上不执行删除操作，而是将对象设置为在下次刷新时删除。该操作还将级联到相关对象，具体取决于我们配置的级联选项，在本例中，级联到相关 Address 对象。

>>> session.delete(patrick)
SELECT user_account.id AS user_account_id, user_account.name AS user_account_name, user_account.fullname AS user_account_fullname
FROM user_account
WHERE user_account.id = ?
[...] (3,)
SELECT address.id AS address_id, address.email_address AS address_email_address, address.user_id AS address_user_id
FROM address
WHERE ? = address.user_id
[...] (3,)

在本例中，Session.delete() 方法发出了两个 SELECT 语句，即使它没有发出 DELETE，这可能令人惊讶。这是因为当该方法检查该对象时，发现 patrick 对象是已过期的，这发生在我们上次调用 Session.commit() 时，并且发出的 SQL 用于从新事务中重新加载行。这种过期是可选的，在正常使用中，我们经常会为不适用的情况关闭它。

为了说明行被删除，这里给出提交操作。

>>> session.commit()
DELETE FROM address WHERE address.id = ?
[...] (4,)
DELETE FROM user_account WHERE user_account.id = ?
[...] (3,)
COMMIT

教程在使用工作单元模式删除 ORM 对象中讨论了 ORM 删除操作。有关对象过期的背景信息，请参阅过期/刷新；有关级联的深入讨论，请参阅级联。

深入了解上述概念¶

对于新用户来说，上面的部分可能是一次快速浏览。上面的每个步骤中都有很多重要的概念没有涵盖。在快速概述了这些概念的外观之后，建议您完成 SQLAlchemy 统一教程，以获得对上述内容的扎实了解。祝您好运！

CryptoQuant量化开源框架

CryptoQuant 量化开源框架