非传统映射
将类映射到多个表
映射器可以构造与任意关系单元（称为 selectables）相对应的类，除了普通表之外。例如，join() 函数创建了一个包含多个表的可选择单元，具有自己的复合主键，可以与 Table 相同的方式映射：
from sqlalchemy import Table, Column, Integer, String, MetaData, join, ForeignKey from sqlalchemy.orm import DeclarativeBase from sqlalchemy.orm import column_property metadata_obj = MetaData() # define two Table objects user_table = Table( "user", metadata_obj, Column("id", Integer, primary_key=True), Column("name", String), ) address_table = Table( "address", metadata_obj, Column("id", Integer, primary_key=True), Column("user_id", Integer, ForeignKey("user.id")), Column("email_address", String), ) # define a join between them. This # takes place across the user.id and address.user_id # columns. user_address_join = join(user_table, address_table) class Base(DeclarativeBase): metadata = metadata_obj # map to it class AddressUser(Base): __table__ = user_address_join id = column_property(user_table.c.id, address_table.c.user_id) address_id = address_table.c.id
在上面的示例中，连接表示了 user 表和 address 表的列。user.id 和 address.user_id 列通过外键相等，因此在映射中它们被定义为一个属性，即 AddressUser.id，使用 column_property() 来指示一个特殊的列映射。基于这部分配置，当发生 flush 时，映射将把新的主键值从 user.id 复制到 address.user_id 列。
另外，address.id 列被显式映射到一个名为 address_id 的属性。这是为了消除歧义，将 address.id 列的映射与同名的 AddressUser.id 属性区分开来，这里已经被分配为引用 user 表与 address.user_id 外键结合的表。
上面映射的自然主键是 (user.id, address.id) 的组合，因为这些是 user 和 address 表的联合主键列。AddressUser 对象的标识将根据这两个值，并且从 AddressUser 对象表示为 (AddressUser.id, AddressUser.address_id)。
当涉及到 AddressUser.id 列时，大多数 SQL 表达式将仅使用映射列列表中的第一列，因为这两列是同义的。然而，对于特殊用例，比如 GROUP BY 表达式，在这种情况下需要同时引用两列，并且在使用正确的上下文时，即考虑到别名和类似情况时，可以使用访问器 Comparator.expressions：
stmt = select(AddressUser).group_by(*AddressUser.id.expressions)
新功能在版本 1.3.17 中添加：增加了 Comparator.expressions 访问器。
注意
如上所示的针对多个表的映射支持持久化，即对目标表中的行进行 INSERT、UPDATE 和 DELETE。然而，它不支持一次为一个记录在一个表上执行 UPDATE 并在其他表上同时执行 INSERT 或 DELETE 的操作。也就是说，如果一个记录 PtoQ 被映射到“p”和“q”表，其中它基于“p”和“q”的 LEFT OUTER JOIN 有一行，如果进行一个 UPDATE 来修改现有记录中“q”表中的数据，那么“q”中的行必须存在；如果主键标识已经存在，它不会发出 INSERT。如果行不存在，对于大多数支持报告 UPDATE 受影响行数的 DBAPI 驱动程序，ORM 将无法检测到更新的行并引发错误；否则，数据将被静默忽略。
一个允许在相关行上“即时”插入的方法可能会使用.MapperEvents.before_update 事件，并且看起来像：
from sqlalchemy import event @event.listens_for(PtoQ, "before_update") def receive_before_update(mapper, connection, target): if target.some_required_attr_on_q is None: connection.execute(q_table.insert(), {"id": target.id})
在上面的例子中，通过使用Table.insert()创建一个 INSERT 构造，然后使用给定的Connection执行它，将一行 INSERT 到q_table表中，这个 Connection 与用于发出 flush 过程中的其他 SQL 的 Connection 相同。用户提供的逻辑必须检测到从“p”到“q”的 LEFT OUTER JOIN 没有“q”侧的条目。## 对任意子查询映射类
类似于针对连接的映射，也可以将一个普通的select()对象与映射器一起使用。下面的示例片段说明了将一个名为Customer的类映射到一个包含与子查询连接的select()中：
from sqlalchemy import select, func subq = ( select( func.count(orders.c.id).label("order_count"), func.max(orders.c.price).label("highest_order"), orders.c.customer_id, ) .group_by(orders.c.customer_id) .subquery() ) customer_select = ( select(customers, subq) .join_from(customers, subq, customers.c.id == subq.c.customer_id) .subquery() ) class Customer(Base): __table__ = customer_select
在上面，由customer_select表示的完整行将是customers表的所有列，以及subq子查询暴露的那些列，即order_count、highest_order和customer_id。将Customer类映射到这个可选择的内容，然后创建一个包含这些属性的类。
当 ORM 持久化Customer的新实例时，实际上只有customers表会收到 INSERT。这是因为orders表的主键没有在映射中表示；ORM 只会对已经映射了主键的表发出 INSERT。
注意
映射到任意 SELECT 语句的做法，特别是像上面这样复杂的语句，几乎从不需要；它必然倾向于生成复杂的查询，这些查询通常比直接构造查询要低效。这种做法在某种程度上基于 SQLAlchemy 的非常早期历史，其中Mapper构造被认为是主要的查询接口；在现代用法中，Query对象可以用于构造几乎任何 SELECT 语句，包括复杂的复合语句，并且应优先于“映射到可选”的方法。
为一个类映射多个映射器
在现代的 SQLAlchemy 中，一个特定的类一次只能由一个所谓的主要映射器（mapper）映射。这个映射器涉及三个主要功能领域：查询、持久性和对映射类的仪器化。主要映射器的理论基础与以下事实相关：Mapper修改了类本身，不仅将其持久化到特定的Table中，还对类上的属性进行了仪器化，这些属性根据表元数据特别结构化。不能有多个映射器与一个类同等相关，因为只有一个映射器可以实际仪器化该类。
“非主要”映射器的概念已经存在多个 SQLAlchemy 版本，但从 1.3 版本开始，此功能已被弃用。其中一个非主要映射器有用的情况是构建与备用可选择类之间的关系时。现在可以使用aliased构造来满足此用例，并在 Relationship to Aliased Class 中进行了描述。
就一个类可以在不同情境下被完全持久化到不同表中的用例而言，早期版本的 SQLAlchemy 提供了一个来自 Hibernate 的功能，称为“实体名称”功能。然而，在 SQLAlchemy 中，一旦映射类本身成为 SQL 表达式构造的来源，即类的属性直接链接到映射表列，这个用例就变得不可行了。该功能被移除，并被一个简单的面向配方的方法取代，以完成此任务而不产生任何仪器化的歧义——创建新的子类，每个类都被单独映射。该模式现在作为一种配方在Entity Name中提供。
将一个类映射到多个表
Mappers 可以针对任意关系单元（称为selectables）进行构建，而不仅仅是普通的表。例如，join() 函数创建了一个包含多个表的可选单元，其中包括其自己的复合主键，可以与Table 以相同的方式映射：
from sqlalchemy import Table, Column, Integer, String, MetaData, join, ForeignKey from sqlalchemy.orm import DeclarativeBase from sqlalchemy.orm import column_property metadata_obj = MetaData() # define two Table objects user_table = Table( "user", metadata_obj, Column("id", Integer, primary_key=True), Column("name", String), ) address_table = Table( "address", metadata_obj, Column("id", Integer, primary_key=True), Column("user_id", Integer, ForeignKey("user.id")), Column("email_address", String), ) # define a join between them. This # takes place across the user.id and address.user_id # columns. user_address_join = join(user_table, address_table) class Base(DeclarativeBase): metadata = metadata_obj # map to it class AddressUser(Base): __table__ = user_address_join id = column_property(user_table.c.id, address_table.c.user_id) address_id = address_table.c.id
在上面的示例中，连接表示了user和address表的列。 user.id和address.user_id列由外键等于，因此在映射中它们被定义为一个属性AddressUser.id，使用column_property()指示专门的列映射。根据配置的这一部分，当发生刷新时，映射将新的主键值从user.id复制到address.user_id列。
另外，address.id列显式映射到名为address_id的属性。这是为了消除歧义，将address.id列的映射与同名的AddressUser.id属性分开，这里已经被分配为引用user表与address.user_id外键的属性。
上述映射的自然主键是(user.id, address.id)的组合，因为这些是user和address表的主键列合并在一起。 AddressUser对象的标识将根据这两个值，并且从AddressUser对象表示为(AddressUser.id, AddressUser.address_id)。
在引用AddressUser.id列时，大多数 SQL 表达式将仅使用映射列列表中的第一列，因为这两列是同义的。但是，对于特殊用例，例如必须同时引用两列的 GROUP BY 表达式，同时考虑到适当的上下文，即适应别名等，可以使用访问器Comparator.expressions：
stmt = select(AddressUser).group_by(*AddressUser.id.expressions)
1.3.17 版本中的新内容：添加了Comparator.expressions 访问器。
注意
如上所示的对多个表的映射支持持久性，即对目标表中的行进行 INSERT、UPDATE 和 DELETE 操作。然而，它不支持在一条记录中同时对一个表进行 UPDATE 并在其他表上执行 INSERT 或 DELETE 的操作。也就是说，如果将记录 PtoQ 映射到“p”和“q”表，其中它基于“p”和“q”的 LEFT OUTER JOIN 的行，如果进行更新以更改现有记录中“q”表中的数据，则“q”中的行必须存在；如果主键标识已经存在，它不会发出 INSERT。如果行不存在，对于大多数支持报告 UPDATE 受影响行数的 DBAPI 驱动程序，ORM 将无法检测到更新的行并引发错误；否则，数据将被静默忽略。
允许在“插入”相关行时使用的配方可能利用.MapperEvents.before_update事件，并且看起来像：
from sqlalchemy import event @event.listens_for(PtoQ, "before_update") def receive_before_update(mapper, connection, target): if target.some_required_attr_on_q is None: connection.execute(q_table.insert(), {"id": target.id})
在上述情况下，通过使用Table.insert()创建一个 INSERT 构造将一行插入q_table表，然后使用给定的Connection执行它，这与用于发出刷新过程中的其他 SQL 的相同连接。用户提供的逻辑必须检测从“p”到“q”的 LEFT OUTER JOIN 是否没有“q”方面的条目。
将类映射到任意子查询
类似于对连接进行映射，也可以将一个普通的select()对象与映射器一起使用。下面的示例片段说明了将名为Customer的类映射到包含与子查询连接的select()的过程：
from sqlalchemy import select, func subq = ( select( func.count(orders.c.id).label("order_count"), func.max(orders.c.price).label("highest_order"), orders.c.customer_id, ) .group_by(orders.c.customer_id) .subquery() ) customer_select = ( select(customers, subq) .join_from(customers, subq, customers.c.id == subq.c.customer_id) .subquery() ) class Customer(Base): __table__ = customer_select
在上面，由customer_select表示的完整行将是customers表的所有列，以及subq子查询暴露的那些列，即order_count、highest_order和customer_id。将Customer类映射到这个可选择的类，然后创建一个包含这些属性的类。
当 ORM 持久化Customer的新实例时，实际上只有customers表会收到 INSERT。这是因为orders表的主键没有在映射中表示；ORM 只会对已映射主键的表发出 INSERT。
注意
对任意 SELECT 语句进行映射的实践，特别是上面那种复杂的情况，几乎从不需要；这必然会产生复杂的查询，通常比直接构造的查询效率低。这种做法在某种程度上基于 SQLAlchemy 的早期历史，其中Mapper构造旨在代表主要的查询接口；在现代用法中，Query对象可用于构造几乎任何 SELECT 语句，包括复杂的复合语句，并且应优先使用“映射到可选择”方法。
一个类对应多个映射器
在现代的 SQLAlchemy 中，一个特定的类在任何时候只被一个所谓的主要映射器所映射。这个映射器涉及三个主要功能领域：查询、持久化和对映射类的仪器化。主要映射器的理念与以下事实相关：Mapper不仅修改类本身，而且将其持久化到特定的Table，还会根据表元数据结构化地仪器化类上的属性。不可能有多个映射器与一个类一样平等地关联，因为只有一个映射器实际上可以仪器化这个类。
“非主要”映射器的概念在许多版本的 SQLAlchemy 中一直存在，但自版本 1.3 起，此功能已不建议使用。唯一需要非主要映射器的情况是在构造与另一个可选择的类的关系时。现在，可以使用aliased构造来满足这个用例，并在关系到别名类中进行描述。
就一个类在不同情境下可以完全持久化到不同表的用例而言，SQLAlchemy 的早期版本提供了一个从 Hibernate 改编而来的功能，称为“实体名称”功能。然而，在 SQLAlchemy 中，一旦映射的类本身成为 SQL 表达式构造的源，即类的属性直接链接到映射表的列，这个用例就变得不可行了。该功能被移除，并用一个简单的基于配方的方法来完成这个任务，而不会有任何仪器化的歧义 - 即创建新的子类，每个类都单独映射。这种模式现在作为实体名称的配方可用。
配置版本计数器
Mapper支持管理版本 ID 列，它是单个表列，每当对映射表进行UPDATE时，该列会递增或以其他方式更新其值。每次 ORM 发出UPDATE或DELETE对行进行操作时，都会检查该值，以确保内存中持有的值与数据库值匹配。
警告
因为版本控制功能依赖于对象的内存记录的比较，所以该功能仅适用于Session.flush()过程，在此过程中 ORM 将单个内存中的行刷新到数据库。当执行多行 UPDATE 或 DELETE 时，该功能不会生效，使用Query.update()或Query.delete()方法，因为这些方法仅发出 UPDATE 或 DELETE 语句，但否则无法直接访问受影响行的内容。
此功能的目的是检测两个并发事务在大致相同的时间修改同一行，或者在可能重用上一个事务的数据而不进行刷新的系统中提供对“过时”行的防护（例如，如果使用Session的expire_on_commit=False设置，可能会重用上一个事务的数据）。
简单版本计数
跟踪版本的最直接方法是向映射表添加一个整数列，然后在映射器选项中将其设为version_id_col：
class User(Base): __tablename__ = "user" id = mapped_column(Integer, primary_key=True) version_id = mapped_column(Integer, nullable=False) name = mapped_column(String(50), nullable=False) __mapper_args__ = {"version_id_col": version_id}
注意
强烈建议将version_id列设为 NOT NULL。版本控制功能不支持版本列中的 NULL 值。
在上面的例子中，User映射使用version_id列跟踪整数版本。当首次刷新User类型的对象时，version_id列的值将为“1”。然后，稍后对表的 UPDATE 将始终以类似以下的方式发出：
UPDATE user SET version_id=:version_id, name=:name WHERE user.id = :user_id AND user.version_id = :user_version_id -- {"name": "new name", "version_id": 2, "user_id": 1, "user_version_id": 1}
上述 UPDATE 语句正在更新不仅与user.id = 1匹配的行，而且还要求user.version_id = 1，其中“1”是我们已知在此对象上使用的最后版本标识符。如果在其他地方的事务独立修改了该行，则此版本 id 将不再匹配，并且 UPDATE 语句将报告没有匹配的行；这是 SQLAlchemy 测试的条件，确保我们的 UPDATE（或 DELETE）语句匹配了恰好一行。如果没有匹配的行，这表明我们的数据版本已过时，并且会引发StaleDataError异常。
自定义版本计数器/类型
可以使用其他类型或计数器来进行版本控制。常见类型包括日期和 GUID。当使用备用类型或计数器方案时，SQLAlchemy 提供了使用version_id_generator参数的钩子，该参数接受一个版本生成可调用对象。此可调用对象会传递当前已知版本的值，并且预期返回后续版本。
例如，如果我们想使用随机生成的 GUID 跟踪User类的版本控制，我们可以这样做（请注意，某些后端支持原生 GUID 类型，但我们在这里使用简单的字符串进行说明）：
import uuid class User(Base): __tablename__ = "user" id = mapped_column(Integer, primary_key=True) version_uuid = mapped_column(String(32), nullable=False) name = mapped_column(String(50), nullable=False) __mapper_args__ = { "version_id_col": version_uuid, "version_id_generator": lambda version: uuid.uuid4().hex, }
每次User对象需要进行 INSERT 或 UPDATE 操作时，持久化引擎将调用uuid.uuid4()。在这种情况下，我们的版本生成函数可以忽略version的传入值，因为uuid4()函数生成的标识符不需要任何先决条件值。如果我们使用的是顺序版本控制方案，例如数字或特殊字符系统，我们可以利用给定的version来帮助确定后续值。
另请参阅
跨后端通用 GUID 类型 ## 服务器端版本计数器
version_id_generator也可以配置为依赖于数据库生成的值。在这种情况下，数据库需要某种方式在行进行 INSERT 时生成新的标识符，以及在 UPDATE 时生成。对于 UPDATE 情况，通常需要一个更新触发器，除非所涉及的数据库支持其他一些本地版本标识符。特别是 PostgreSQL 数据库支持一个称为xmin的系统列，它提供了 UPDATE 版本控制。我们可以如下使用 PostgreSQL 的xmin列为我们的User类版本控制：
from sqlalchemy import FetchedValue class User(Base): __tablename__ = "user" id = mapped_column(Integer, primary_key=True) name = mapped_column(String(50), nullable=False) xmin = mapped_column("xmin", String, system=True, server_default=FetchedValue()) __mapper_args__ = {"version_id_col": xmin, "version_id_generator": False}
使用上述映射，ORM 将依赖于xmin列自动提供版本 id 计数器的新值。
当 ORM 发出 INSERT 或 UPDATE 时，通常不会主动获取数据库生成的值，而是将这些列保留为“过期”，在下次访问它们时获取，除非设置了eager_defaults Mapper标志。但是，当使用服务器端版本列时，ORM 需要主动获取新生成的值。这样做是为了在任何并发事务可能再次更新之前设置版本计数器之前。最好同时在 INSERT 或 UPDATE 语句中使用 RETURNING 进行获取，否则，如果之后发出 SELECT 语句，则仍然存在潜在的竞争条件，版本计数器可能在获取之前更改。
当目标数据库支持 RETURNING 时，我们的User类的 INSERT 语句如下所示：
INSERT INTO "user" (name) VALUES (%(name)s) RETURNING "user".id, "user".xmin -- {'name': 'ed'}
在上述情况下，ORM 可以在一条语句中获取任何新生成的主键值以及服务器生成的版本标识符。当后端不支持 RETURNING 时，必须对每个INSERT 和 UPDATE 发出额外的 SELECT，这非常低效，还会引入可能丢失版本计数器的可能性：
INSERT INTO "user" (name) VALUES (%(name)s) -- {'name': 'ed'} SELECT "user".version_id AS user_version_id FROM "user" where "user".id = :param_1 -- {"param_1": 1}
强烈建议仅在绝对必要时且仅在支持 RETURNING 的后端上使用服务器端版本计数器，目前支持的后端有 PostgreSQL、Oracle、MariaDB 10.5、SQLite 3.35 和 SQL Server。
SqlAlchemy 2.0 中文文档（十）（2）https://developer.aliyun.com/article/1562916
