ORM 快速入门¶

对于想要快速了解基本 ORM 用法的新用户，这里是 SQLAlchemy 统一教程中使用的映射和示例的缩略形式。这里的代码可以从干净的命令行完全运行。

由于本节中的描述有意非常简短，请继续阅读完整的 SQLAlchemy 统一教程，以获得对此处说明的每个概念的更深入的描述。

版本 2.0 中的更改: ORM 快速入门已针对最新的 PEP 484 感知特性进行了更新，使用了包括 mapped_column() 在内的新构造。有关迁移信息，请参阅 ORM 声明式模型部分。

声明模型¶

在这里，我们定义模块级构造，这些构造将形成我们将从中查询数据库的结构。此结构称为声明式映射，它同时定义了 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 映射类都要求至少将一列声明为主键的一部分，通常通过在应成为键一部分的那些 mapped_column() 对象上使用 Column.primary_key 参数来实现。在上面的示例中，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__() 之类的方法来创建映射类。有关数据类映射的更多信息，请参阅声明式数据类映射。

创建引擎¶

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，事务就始终在进行中

>>> fromsqlalchemy.ormimportSession>>> withSession(engine)assession:...     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 Session 执行中，更多信息在使用 Session 的基础知识中。

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

简单 SELECT¶

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

>>> fromsqlalchemyimportselect>>> session=Session(engine)>>> stmt=select(User).where(User.name.in_(["spongebob","sandy"]))>>> foruserinsession.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_addressAddress(id=2, email_address='sandy@sqlalchemy.org')

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

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

进行更改¶

Session 对象与我们的 ORM 映射类 User 和 Address 结合使用，会自动跟踪对对象所做的更改，这些更改会导致 SQL 语句在下次 Session 刷新时发出。下面，我们更改与“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 统一教程，以获得对上面真正发生的事情的扎实工作知识。祝你好运！

SQLAlchemy 2.0 文档