relationship()
通常通过检查两个表之间的外键关系来创建两个表之间的联接,以确定应该比较哪些列。在各种情况下,需要对这种行为进行定制。
要处理的最常见情况之一是两个表之间有多个外键路径。
考虑一下 Customer
类的两个外键 Address
班级:
from sqlalchemy import Integer, ForeignKey, String, Column
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
Base = declarative_base()
class Customer(Base):
__tablename__ = 'customer'
id = Column(Integer, primary_key=True)
name = Column(String)
billing_address_id = Column(Integer, ForeignKey("address.id"))
shipping_address_id = Column(Integer, ForeignKey("address.id"))
billing_address = relationship("Address")
shipping_address = relationship("Address")
class Address(Base):
__tablename__ = 'address'
id = Column(Integer, primary_key=True)
street = Column(String)
city = Column(String)
state = Column(String)
zip = Column(String)
当我们尝试使用上面的映射时,会产生以下错误:
sqlalchemy.exc.AmbiguousForeignKeysError: Could not determine join
condition between parent/child tables on relationship
Customer.billing_address - there are multiple foreign key
paths linking the tables. Specify the 'foreign_keys' argument,
providing a list of those columns which should be
counted as containing a foreign key reference to the parent table.
上面的信息很长。有许多潜在的信息 relationship()
can-return,它经过精心定制,可以检测到各种常见的配置问题;大多数会建议解决歧义或其他丢失信息所需的额外配置。
在这种情况下,消息希望我们限定每个 relationship()
通过指示每一列应考虑哪个外键列,相应的格式如下:
class Customer(Base):
__tablename__ = 'customer'
id = Column(Integer, primary_key=True)
name = Column(String)
billing_address_id = Column(Integer, ForeignKey("address.id"))
shipping_address_id = Column(Integer, ForeignKey("address.id"))
billing_address = relationship("Address", foreign_keys=[billing_address_id])
shipping_address = relationship("Address", foreign_keys=[shipping_address_id])
上面,我们指定了 foreign_keys
参数,它是 Column
或列表 Column
对象,指示那些列被视为“外部”,或者换句话说,包含引用父表的值的列。加载 Customer.billing_address
与A的关系 Customer
对象将使用 billing_address_id
以便识别中的行 Address
装载;类似地, shipping_address_id
用于 shipping_address
关系。两列的链接在持久化过程中也起作用;刚插入的 Address
对象将被复制到关联的 Customer
刷新过程中的对象。
指定时 foreign_keys
对于声明性的,我们也可以使用字符串名称来指定,但是,如果使用列表,则 列表是字符串的一部分 ::
billing_address = relationship("Address", foreign_keys="[Customer.billing_address_id]")
在这个特定的示例中,在任何情况下都不需要列表,因为只有一个列表 Column
我们需要:
billing_address = relationship("Address", foreign_keys="Customer.billing_address_id")
的默认行为 relationship()
在构造联接时,它将一边的主键列的值与另一边的外键引用列的值相等。我们可以将此标准更改为任何我们想使用的 primaryjoin
争论,以及 secondaryjoin
使用“secondary”表时的参数。
在下面的示例中,使用 User
类以及 Address
类存储街道地址,我们创建关系 boston_addresses
只装那些 Address
指定“波士顿”城市的对象:
from sqlalchemy import Integer, ForeignKey, String, Column
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
Base = declarative_base()
class User(Base):
__tablename__ = 'user'
id = Column(Integer, primary_key=True)
name = Column(String)
boston_addresses = relationship("Address",
primaryjoin="and_(User.id==Address.user_id, "
"Address.city=='Boston')")
class Address(Base):
__tablename__ = 'address'
id = Column(Integer, primary_key=True)
user_id = Column(Integer, ForeignKey('user.id'))
street = Column(String)
city = Column(String)
state = Column(String)
zip = Column(String)
在这个字符串SQL表达式中,我们使用了 and_()
连接构造为连接条件建立两个不同的谓词-连接 User.id
和 Address.user_id
列之间以及限制中的行 Address
只是 city='Boston'
. 当使用声明性的、基本的SQL函数时,例如 and_()
在字符串的计算命名空间中自动可用 relationship()
争论。
我们在 primaryjoin
通常只有当SQLAlchemy为了加载或表示此关系而呈现SQL时才有意义。也就是说,它用于发出的SQL语句中,以便执行每个属性的惰性加载,或者在查询时构造联接,例如通过 Query.join()
或者通过“联接”或“子查询”的加载样式。当内存中的对象被操纵时,我们可以放置任何 Address
我们想进入的对象 boston_addresses
集合,无论 .city
属性为。对象将一直存在于集合中,直到属性过期并从应用条件的数据库重新加载。当刷新发生时,对象在 boston_addresses
将无条件刷新,分配主键的值 user.id
列到外键保持上 address.user_id
每行的列。这个 city
条件在这里不起作用,因为刷新过程只关心将主键值同步到引用外键值。
主连接条件的另一个元素是如何确定那些被认为是“外部”的列。通常是 Column
对象将指定 ForeignKey
或以其他方式成为 ForeignKeyConstraint
这与连接条件有关。 relationship()
查找此外键状态,因为它决定如何为该关系加载和持久化数据。然而, primaryjoin
参数可用于创建不涉及任何“架构”级外键的联接条件。我们可以结合 primaryjoin
随着 foreign_keys
和 remote_side
为了建立这样的连接。
下面,一个班 HostEntry
连接到自身,等于字符串 content
列到 ip_address
列,它是PostgreSQL类型,名为 INET
. 我们需要使用 cast()
为了将连接的一侧强制转换为另一侧的类型:
from sqlalchemy import cast, String, Column, Integer
from sqlalchemy.orm import relationship
from sqlalchemy.dialects.postgresql import INET
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class HostEntry(Base):
__tablename__ = 'host_entry'
id = Column(Integer, primary_key=True)
ip_address = Column(INET)
content = Column(String(50))
# relationship() using explicit foreign_keys, remote_side
parent_host = relationship("HostEntry",
primaryjoin=ip_address == cast(content, INET),
foreign_keys=content,
remote_side=ip_address
)
上述关系将产生一个连接,如:
SELECT host_entry.id, host_entry.ip_address, host_entry.content
FROM host_entry JOIN host_entry AS host_entry_1
ON host_entry_1.ip_address = CAST(host_entry.content AS INET)
上面的另一种语法是使用 foreign()
和 remote()
annotations ,在 primaryjoin
表达式。此语法表示 relationship()
通常单独应用于给定 foreign_keys
和 remote_side
争论。当存在显式联接条件时,这些函数可能更简洁,而且还可以精确地标记“外部”或“远程”列,而不管该列是多次声明还是在复杂的SQL表达式中声明:
from sqlalchemy.orm import foreign, remote
class HostEntry(Base):
__tablename__ = 'host_entry'
id = Column(Integer, primary_key=True)
ip_address = Column(INET)
content = Column(String(50))
# relationship() using explicit foreign() and remote() annotations
# in lieu of separate arguments
parent_host = relationship("HostEntry",
primaryjoin=remote(ip_address) == \
cast(foreign(content), INET),
)
关系的另一个用例是使用自定义运算符,例如PostgreSQL的“包含在” <<
与类型(如)联接时的运算符 postgresql.INET
和 postgresql.CIDR
. 对于自定义运算符,我们使用 Operators.op()
功能:
inet_column.op("<<")(cidr_column)
但是,如果我们构造一个 primaryjoin
使用这个运算符, relationship()
仍然需要更多信息。这是因为当它检查我们的PrimaryJoin条件时,它专门查找用于 比较 ,这通常是包含已知比较运算符的固定列表,例如 ==
, <
等。因此,为了让我们的自定义运算符参与此系统,我们需要它使用 is_comparison
参数::
inet_column.op("<<", is_comparison=True)(cidr_column)
一个完整的例子:
class IPA(Base):
__tablename__ = 'ip_address'
id = Column(Integer, primary_key=True)
v4address = Column(INET)
network = relationship("Network",
primaryjoin="IPA.v4address.op('<<', is_comparison=True)"
"(foreign(Network.v4representation))",
viewonly=True
)
class Network(Base):
__tablename__ = 'network'
id = Column(Integer, primary_key=True)
v4representation = Column(CIDR)
上面的查询,例如:
session.query(IPA).join(IPA.network)
将呈现为:
SELECT ip_address.id AS ip_address_id, ip_address.v4address AS ip_address_v4address
FROM ip_address JOIN network ON ip_address.v4address << network.v4representation
0.9.2 新版功能: -增加了 Operators.op.is_comparison
用于协助创建的标志 relationship()
使用自定义运算符构造。
当使用组合外键时,可能会出现一种罕见的情况,例如单个列可能是通过外键约束引用的多个列的主题。
考虑(公认的复杂)映射,例如 Magazine
对象,由 Writer
对象与 Article
使用复合主键方案的对象,该方案包括 magazine_id
两者都有;然后 Article
参照 Writer
也, Article.magazine_id
参与两个独立的关系; Article.magazine
和 Article.writer
::
class Magazine(Base):
__tablename__ = 'magazine'
id = Column(Integer, primary_key=True)
class Article(Base):
__tablename__ = 'article'
article_id = Column(Integer)
magazine_id = Column(ForeignKey('magazine.id'))
writer_id = Column()
magazine = relationship("Magazine")
writer = relationship("Writer")
__table_args__ = (
PrimaryKeyConstraint('article_id', 'magazine_id'),
ForeignKeyConstraint(
['writer_id', 'magazine_id'],
['writer.id', 'writer.magazine_id']
),
)
class Writer(Base):
__tablename__ = 'writer'
id = Column(Integer, primary_key=True)
magazine_id = Column(ForeignKey('magazine.id'), primary_key=True)
magazine = relationship("Magazine")
配置上述映射后,我们将看到此警告发出::
SAWarning: relationship 'Article.writer' will copy column
writer.magazine_id to column article.magazine_id,
which conflicts with relationship(s): 'Article.magazine'
(copies magazine.id to article.magazine_id). Consider applying
viewonly=True to read-only relationships, or provide a primaryjoin
condition marking writable columns with the foreign() annotation.
这所指的源于以下事实: Article.magazine_id
是两个不同的外键约束的主题;它指 Magazine.id
直接作为源列,但也指 Writer.magazine_id
作为复合键上下文中的源列 Writer
. 如果我们将 Article
有一个特别的 Magazine
,然后将 Article
用一个 Writer
这与 不同的 Magazine
,ORM将覆盖 Article.magazine_id
非确定性地,无声地改变我们所指的杂志;如果我们取消关联,它也可能试图将空值放入此列中。 Writer
从一个 Article
. 警告让我们知道这是事实。
为了解决这个问题,我们需要打破 Article
包括以下三个功能:
Article
首先写给 Article.magazine_id
基于 Article.magazine
仅关系,即从复制的值 Magazine.id
.
Article
可以写信给 Article.writer_id
代表保留在 Article.writer
关系,但只有 Writer.id
列; Writer.magazine_id
列不应写入 Article.magazine_id
因为它最终来源于 Magazine.id
.
Article
拿 Article.magazine_id
加载时计入 Article.writer
尽管如此 不 代表这个关系写信给它。
为了得到1和2,我们只能指定 Article.writer_id
作为“外国钥匙” Article.writer
::
class Article(Base):
# ...
writer = relationship("Writer", foreign_keys='Article.writer_id')
然而,这具有 Article.writer
不采取 Article.magazine_id
查询时转入账户 Writer
:
SELECT article.article_id AS article_article_id,
article.magazine_id AS article_magazine_id,
article.writer_id AS article_writer_id
FROM article
JOIN writer ON writer.id = article.writer_id
因此,为了得到1、2和3的所有内容,我们表示连接条件以及要通过组合写入的列 primaryjoin
完全,以及 foreign_keys
参数,或者更简洁地通过注释 foreign()
::
class Article(Base):
# ...
writer = relationship(
"Writer",
primaryjoin="and_(Writer.id == foreign(Article.writer_id), "
"Writer.magazine_id == Article.magazine_id)")
在 1.0.0 版更改: 当一列同时用作多个关系的同步目标时,ORM将尝试发出警告。
警告
本节详细介绍了一个实验特性。
使用自定义表达式意味着我们可以生成不符合常规主键/外键模型的非常规联接条件。一个这样的例子是物化路径模式,我们在其中比较字符串与重叠的路径标记,以生成树结构。
通过小心使用 foreign()
和 remote()
我们可以建立一种关系,有效地产生一个基本的物化路径系统。基本上,当 foreign()
和 remote()
是在 same 在比较表达式的一边,关系被认为是“一对多”;当它们处于 不同的 双方的关系被认为是“多对一”。为了在这里进行比较,我们将处理集合,因此我们将事物配置为“一对多”::
class Element(Base):
__tablename__ = 'element'
path = Column(String, primary_key=True)
descendants = relationship('Element',
primaryjoin=
remote(foreign(path)).like(
path.concat('/%')),
viewonly=True,
order_by=path)
上面,如果给出 Element
路径属性为的对象 "/foo/bar2"
我们要找一堆 Element.descendants
看起来像::
SELECT element.path AS element_path
FROM element
WHERE element.path LIKE ('/foo/bar2' || '/%') ORDER BY element.path
0.9.5 新版功能: 添加了支持以允许在PrimaryJoin条件内对自身进行单列比较,以及对使用的PrimaryJoin条件进行比较。 ColumnOperators.like()
作为比较运算符。
多对多关系可以由一个或两个 primaryjoin
和 secondaryjoin
-后者对于使用 secondary
争论。涉及使用 primaryjoin
和 secondaryjoin
当从一个类到它自己建立一个多对多关系时,如下所示:
from sqlalchemy import Integer, ForeignKey, String, Column, Table
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
Base = declarative_base()
node_to_node = Table("node_to_node", Base.metadata,
Column("left_node_id", Integer, ForeignKey("node.id"), primary_key=True),
Column("right_node_id", Integer, ForeignKey("node.id"), primary_key=True)
)
class Node(Base):
__tablename__ = 'node'
id = Column(Integer, primary_key=True)
label = Column(String)
right_nodes = relationship("Node",
secondary=node_to_node,
primaryjoin=id==node_to_node.c.left_node_id,
secondaryjoin=id==node_to_node.c.right_node_id,
backref="left_nodes"
)
在上面的位置,SQLAlchemy无法自动知道哪些列应该连接到 right_nodes
和 left_nodes
关系。这个 primaryjoin
和 secondaryjoin
参数确定我们希望如何加入关联表。在上面的声明形式中,正如我们在对应于 Node
类 id
变量直接作为 Column
我们希望加入的对象。
或者,我们可以定义 primaryjoin
和 secondaryjoin
使用字符串的参数,这适用于我们的配置没有 Node.id
列对象可用,或者 node_to_node
可能还没有表格。指平原时 Table
对象在声明性字符串中,我们使用表的字符串名称,因为它存在于 MetaData
::
class Node(Base):
__tablename__ = 'node'
id = Column(Integer, primary_key=True)
label = Column(String)
right_nodes = relationship("Node",
secondary="node_to_node",
primaryjoin="Node.id==node_to_node.c.left_node_id",
secondaryjoin="Node.id==node_to_node.c.right_node_id",
backref="left_nodes"
)
这里的经典映射情况类似,其中 node_to_node
可以加入到 node.c.id
::
from sqlalchemy import Integer, ForeignKey, String, Column, Table, MetaData
from sqlalchemy.orm import relationship, mapper
metadata = MetaData()
node_to_node = Table("node_to_node", metadata,
Column("left_node_id", Integer, ForeignKey("node.id"), primary_key=True),
Column("right_node_id", Integer, ForeignKey("node.id"), primary_key=True)
)
node = Table("node", metadata,
Column('id', Integer, primary_key=True),
Column('label', String)
)
class Node(object):
pass
mapper(Node, node, properties={
'right_nodes':relationship(Node,
secondary=node_to_node,
primaryjoin=node.c.id==node_to_node.c.left_node_id,
secondaryjoin=node.c.id==node_to_node.c.right_node_id,
backref="left_nodes"
)})
注意,在这两个例子中, backref
关键字指定 left_nodes
后退-当 relationship()
以相反的方向创建第二个关系,它足够智能地反转 primaryjoin
和 secondaryjoin
争论。
注解
本节介绍了SQLAlchemy的一些新的和实验性的特性。
有时,当一个人试图建立一个 relationship()
在两个表之间,需要涉及两个或三个以上的表才能联接它们。这是一片 relationship()
在这里,我们试图突破可能的边界,并且通常需要在SQLAlchemy邮件列表中详细说明这些外来用例的最终解决方案。
在最新版本的SQLAlchemy中, secondary
在某些情况下,可以使用参数来提供由多个表组成的复合目标。下面是这样一个连接条件的示例(至少需要0.9.2版才能按原样工作)::
class A(Base):
__tablename__ = 'a'
id = Column(Integer, primary_key=True)
b_id = Column(ForeignKey('b.id'))
d = relationship("D",
secondary="join(B, D, B.d_id == D.id)."
"join(C, C.d_id == D.id)",
primaryjoin="and_(A.b_id == B.id, A.id == C.a_id)",
secondaryjoin="D.id == B.d_id",
uselist=False
)
class B(Base):
__tablename__ = 'b'
id = Column(Integer, primary_key=True)
d_id = Column(ForeignKey('d.id'))
class C(Base):
__tablename__ = 'c'
id = Column(Integer, primary_key=True)
a_id = Column(ForeignKey('a.id'))
d_id = Column(ForeignKey('d.id'))
class D(Base):
__tablename__ = 'd'
id = Column(Integer, primary_key=True)
在上面的示例中,我们提供了 secondary
, primaryjoin
和 secondaryjoin
,以引用命名表的声明性样式 a
, b
, c
, d
直接。从查询 A
到 D
看起来像:
sess.query(A).join(A.d).all()
SELECT a.id AS a_id, a.b_id AS a_b_id
FROM a JOIN (
b AS b_1 JOIN d AS d_1 ON b_1.d_id = d_1.id
JOIN c AS c_1 ON c_1.d_id = d_1.id)
ON a.b_id = b_1.id AND a.id = c_1.a_id JOIN d ON d.id = b_1.d_id
在上面的示例中,我们利用能够将多个表填充到一个“辅助”容器中的优势,这样我们就可以跨多个表进行连接,同时还可以为 relationship()
在这一点上,“左”和“右”都只有“一”个表;复杂性保持在中间。
1.3 新版功能: 这个 AliasedClass
现在可以将构造指定为 relationship()
替换了以前使用非主映射器的方法,这种方法有一些限制,例如它们不继承被映射实体的子关系,而且它们需要针对可选对象进行复杂配置。本节中的方法现在更新为使用 AliasedClass
.
在前一节中,我们演示了一种我们使用的技术 secondary
以便在联接条件中放置其他表。有一种复杂的连接情况,即使这种技术也不够;当我们试图从 A
到 B
,利用任何数量的 C
, D
等,但也有连接条件 A
和 B
直接地 . 在这种情况下,联接来自 A
到 B
可能很难用一个复杂的词来表达 primaryjoin
条件,因为中间表可能需要特殊处理,而且它也不能用 secondary
对象,因为 A->secondary->B
模式不支持之间的任何引用 A
和 B
直接。当这 非常先进 在这种情况下,我们可以通过创建第二个映射作为关系的目标。这是我们使用的地方 AliasedClass
为了映射到一个类,该类包含了这个联接所需的所有附加表。为了生成这个映射器作为类的“可选”映射,我们使用 aliased()
函数生成新构造,然后使用 relationship()
对象,就好像它是一个普通的映射类。
下图说明了 relationship()
通过简单的联接 A
到 B
但是,PrimaryJoin条件被另外两个实体所增强。 C
和 D
,其中也必须有与两行中的行对齐的行 A
和 B
同时:
class A(Base):
__tablename__ = 'a'
id = Column(Integer, primary_key=True)
b_id = Column(ForeignKey('b.id'))
class B(Base):
__tablename__ = 'b'
id = Column(Integer, primary_key=True)
class C(Base):
__tablename__ = 'c'
id = Column(Integer, primary_key=True)
a_id = Column(ForeignKey('a.id'))
class D(Base):
__tablename__ = 'd'
id = Column(Integer, primary_key=True)
c_id = Column(ForeignKey('c.id'))
b_id = Column(ForeignKey('b.id'))
# 1. set up the join() as a variable, so we can refer
# to it in the mapping multiple times.
j = join(B, D, D.b_id == B.id).join(C, C.id == D.c_id)
# 2. Create an AliasedClass to B
B_viacd = aliased(B, j, flat=True)
A.b = relationship(B_viacd, primaryjoin=A.b_id == j.c.b_id)
通过上面的映射,一个简单的连接看起来像:
sess.query(A).join(A.b).all()
SELECT a.id AS a_id, a.b_id AS a_b_id
FROM a JOIN (b JOIN d ON d.b_id = b.id JOIN c ON c.id = d.c_id) ON a.b_id = b.id
关系的另一个有趣的用例 AliasedClass
对象是关系需要连接到任何形式的专门选择的情况。一种情况是需要使用window函数时,例如限制关系应返回多少行。下面的示例说明了将为每个集合加载前十个项的非主映射器关系:
class A(Base):
__tablename__ = 'a'
id = Column(Integer, primary_key=True)
class B(Base):
__tablename__ = 'b'
id = Column(Integer, primary_key=True)
a_id = Column(ForeignKey("a.id"))
partition = select([
B,
func.row_number().over(
order_by=B.id, partition_by=B.a_id
).label('index')
]).alias()
partitioned_b = aliased(B, partition)
A.partitioned_bs = relationship(
partitioned_b,
primaryjoin=and_(partitioned_b.a_id == A.id, partition.c.index < 10)
)
我们可以用上面的 partitioned_bs
与大多数装载机策略的关系,例如 selectinload()
::
for a1 in s.query(A).options(selectinload(A.partitioned_bs)):
print(a1.partitioned_bs) # <-- will be no more than ten objects
上面的“selectinload”查询如下所示:
SELECT
a_1.id AS a_1_id, anon_1.id AS anon_1_id, anon_1.a_id AS anon_1_a_id,
anon_1.data AS anon_1_data, anon_1.index AS anon_1_index
FROM a AS a_1
JOIN (
SELECT b.id AS id, b.a_id AS a_id, b.data AS data,
row_number() OVER (PARTITION BY b.a_id ORDER BY b.id) AS index
FROM b) AS anon_1
ON anon_1.a_id = a_1.id AND anon_1.index < %(index_1)s
WHERE a_1.id IN ( ... primary key collection ...)
ORDER BY a_1.id
上面,对于“a”中的每个匹配主键,我们将按照“b.id”的顺序获得前十个“bs”。通过对“a_id”进行分区,我们确保每个“row number”都是父“a_id”的本地行。
这种映射通常还包括从“a”到“b”的“普通”关系,用于持久性操作,以及当需要每个“a”的完整“b”对象集时。
非常雄心勃勃的自定义连接条件可能无法直接持久化,在某些情况下甚至可能无法正确加载。要删除公式的持久性部分,请使用标志 viewonly
上 relationship()
,它将其建立为只读属性(在flush()上写入集合的数据将被忽略)。但是,在极端情况下,考虑将常规的python属性与 Query
如下:
class User(Base):
__tablename__ = 'user'
id = Column(Integer, primary_key=True)
def _get_addresses(self):
return object_session(self).query(Address).with_parent(self).filter(...).all()
addresses = property(_get_addresses)