Настройка способа присоединения отношений¶

:func:`_orm.relationship`<

Работа с несколькими путями присоединения¶

Одна из наиболее распространенных ситуаций, с которой приходится сталкиваться, - это когда между двумя таблицами существует более одного пути внешнего ключа.

Customer Address C

from sqlalchemy import Integer, ForeignKey, String, Column
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import relationship


class Base(DeclarativeBase):
    pass


class Customer(Base):
    __tablename__ = "customer"
    id = mapped_column(Integer, primary_key=True)
    name = mapped_column(String)

    billing_address_id = mapped_column(Integer, ForeignKey("address.id"))
    shipping_address_id = mapped_column(Integer, ForeignKey("address.id"))

    billing_address = relationship("Address")
    shipping_address = relationship("Address")


class Address(Base):
    __tablename__ = "address"
    id = mapped_column(Integer, primary_key=True)
    street = mapped_column(String)
    city = mapped_column(String)
    state = mapped_column(String)
    zip = mapped_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.

:func:`_orm.relationship`T

:func:`_orm.relationship`I

class Customer(Base):
    __tablename__ = "customer"
    id = mapped_column(Integer, primary_key=True)
    name = mapped_column(String)

    billing_address_id = mapped_column(Integer, ForeignKey("address.id"))
    shipping_address_id = mapped_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 Customer billing_address_id Address shipping_address_id shipping_address Address Customer A

``foreign_keys``W

billing_address = relationship("Address", foreign_keys="[Customer.billing_address_id]")

:class:`_schema.Column`I

billing_address = relationship("Address", foreign_keys="Customer.billing_address_id")

Предупреждение

relationship.foreign_keys eval() Оценка аргументов в пользу отношений relationship() W

Указание альтернативных условий присоединения¶

relationship() relationship.primaryjoin relationship.secondaryjoin T

User Address boston_addresses Address I

from sqlalchemy import Integer, ForeignKey, String, Column
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import relationship


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "user"
    id = mapped_column(Integer, primary_key=True)
    name = mapped_column(String)
    boston_addresses = relationship(
        "Address",
        primaryjoin="and_(User.id==Address.user_id, " "Address.city=='Boston')",
    )


class Address(Base):
    __tablename__ = "address"
    id = mapped_column(Integer, primary_key=True)
    user_id = mapped_column(Integer, ForeignKey("user.id"))

    street = mapped_column(String)
    city = mapped_column(String)
    state = mapped_column(String)
    zip = mapped_column(String)

and_() User.id Address.user_id Address city='Boston' and_() relationship() W

Предупреждение

relationship.primaryjoin eval() Оценка аргументов в пользу отношений relationship() W

relationship.primaryjoin Select.join() Address boston_addresses .city boston_addresses user.id address.user_id city T

Создание пользовательских иностранных условий¶

Column ForeignKey ForeignKeyConstraint relationship() relationship.primaryjoin relationship.primaryjoin relationship.foreign_keys relationship.remote_side A

HostEntry content ip_address INET cast() B

from sqlalchemy import cast, String, Column, Integer
from sqlalchemy.orm import relationship
from sqlalchemy.dialects.postgresql import INET

from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass


class HostEntry(Base):
    __tablename__ = "host_entry"

    id = mapped_column(Integer, primary_key=True)
    ip_address = mapped_column(INET)
    content = mapped_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 relationship.primaryjoin relationship() relationship.foreign_keys relationship.remote_side A

from sqlalchemy.orm import foreign, remote


class HostEntry(Base):
    __tablename__ = "host_entry"

    id = mapped_column(Integer, primary_key=True)
    ip_address = mapped_column(INET)
    content = mapped_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),
    )

Использование пользовательских операторов в условиях присоединения¶

<< INET CIDR Operators.bool_op() A

inet_column.bool_op("<<")(cidr_column)

relationship.primaryjoin relationship() A

class IPA(Base):
    __tablename__ = "ip_address"

    id = mapped_column(Integer, primary_key=True)
    v4address = mapped_column(INET)

    network = relationship(
        "Network",
        primaryjoin="IPA.v4address.bool_op('<<')" "(foreign(Network.v4representation))",
        viewonly=True,
    )


class Network(Base):
    __tablename__ = "network"

    id = mapped_column(Integer, primary_key=True)
    v4representation = mapped_column(CIDR)

Выше приведен запрос типа:

select(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

Пользовательские операторы на основе функций SQL¶

Operators.op.is_comparison FunctionElement.as_comparison() func Geoalchemy2 A

from geoalchemy2 import Geometry
from sqlalchemy import Column, Integer, func
from sqlalchemy.orm import relationship, foreign


class Polygon(Base):
    __tablename__ = "polygon"
    id = mapped_column(Integer, primary_key=True)
    geom = mapped_column(Geometry("POLYGON", srid=4326))
    points = relationship(
        "Point",
        primaryjoin="func.ST_Contains(foreign(Polygon.geom), Point.geom).as_comparison(1, 2)",
        viewonly=True,
    )


class Point(Base):
    __tablename__ = "point"
    id = mapped_column(Integer, primary_key=True)
    geom = mapped_column(Geometry("POINT", srid=4326))

FunctionElement.as_comparison() func.ST_Contains() Polygon.geom Point.geom foreign() A

Добавлено в версии 1.3: :meth:`.FunctionElement.as_comparison`A

Перекрывающиеся внешние ключи¶

Редкий сценарий может возникнуть при использовании составных внешних ключей, когда один столбец может быть объектом более чем одного столбца, на который ссылается ограничение внешнего ключа.

Magazine Writer Article magazine_id Article Writer Article.magazine_id Article.magazine Article.writer C

class Magazine(Base):
    __tablename__ = "magazine"

    id = mapped_column(Integer, primary_key=True)


class Article(Base):
    __tablename__ = "article"

    article_id = mapped_column(Integer)
    magazine_id = mapped_column(ForeignKey("magazine.id"))
    writer_id = mapped_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 = mapped_column(Integer, primary_key=True)
    magazine_id = mapped_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 Article.magazine_id Writer Article W

``Article``T

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 <

Article.writer_id Article.writer T

class Article(Base):
    # ...

    writer = relationship("Writer", foreign_keys="Article.writer_id")

Article.writer Article.magazine_id Writer H

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

relationship.primaryjoin relationship.foreign_keys foreign() T

class Article(Base):
    # ...

    writer = relationship(
        "Writer",
        primaryjoin="and_(Writer.id == foreign(Article.writer_id), "
        "Writer.magazine_id == Article.magazine_id)",
    )

Нереляционные сравнения / Материализованный путь¶

Предупреждение

в этом разделе описывается экспериментальная функция.

Использование пользовательских выражений позволяет создавать нестандартные условия присоединения, которые не подчиняются обычной модели первичного/отдельного ключа. Одним из таких примеров является шаблон материализованного пути, в котором мы сравниваем строки на предмет пересекающихся лексем пути, чтобы получить древовидную структуру.

foreign() remote() foreign() remote() T

class Element(Base):
    __tablename__ = "element"

    path = mapped_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 A

SELECT element.path AS element_path
FROM element
WHERE element.path LIKE ('/foo/bar2' || '/%') ORDER BY element.path

Отношения «многие-ко-многим» с самореференцией¶

См.также

Отношения в списке смежности Стратегии самореференциальных запросов Настройка самовоспроизводящейся ускоренной загрузки T

relationship.primaryjoin relationship.secondaryjoin relationship.secondary relationship.primaryjoin relationship.secondaryjoin M

from sqlalchemy import Integer, ForeignKey, String, Column, Table
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import relationship


class Base(DeclarativeBase):
    pass


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 = mapped_column(Integer, primary_key=True)
    label = mapped_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",
    )

right_nodes left_nodes relationship.primaryjoin relationship.secondaryjoin Node id Column W

relationship.primaryjoin relationship.secondaryjoin Node.id node_to_node Table MetaData A

class Node(Base):
    __tablename__ = "node"
    id = mapped_column(Integer, primary_key=True)
    label = mapped_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",
    )

Предупреждение

relationship.primaryjoin relationship.secondaryjoin eval() Оценка аргументов в пользу отношений relationship() W

node_to_node node.c.id A

from sqlalchemy import Integer, ForeignKey, String, Column, Table, MetaData
from sqlalchemy.orm import relationship, registry

metadata_obj = MetaData()
mapper_registry = registry()

node_to_node = Table(
    "node_to_node",
    metadata_obj,
    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_obj,
    Column("id", Integer, primary_key=True),
    Column("label", String),
)


class Node:
    pass


mapper_registry.map_imperatively(
    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",
        )
    },
)

relationship.backref left_nodes relationship() relationship.primaryjoin relationship.secondaryjoin N

См.также

:ref:`self_referential`<
:ref:`self_referential_query`<
:ref:`self_referential_eager_loading`<

Составные «вторичные» соединения¶

Примечание

:ref:`in-Python attributes <mapper_hybrids>`T

relationship() relationship() S

:paramref:`_orm.relationship.secondary`I

class A(Base):
    __tablename__ = "a"

    id = mapped_column(Integer, primary_key=True)
    b_id = mapped_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,
        viewonly=True,
    )


class B(Base):
    __tablename__ = "b"

    id = mapped_column(Integer, primary_key=True)
    d_id = mapped_column(ForeignKey("d.id"))


class C(Base):
    __tablename__ = "c"

    id = mapped_column(Integer, primary_key=True)
    a_id = mapped_column(ForeignKey("a.id"))
    d_id = mapped_column(ForeignKey("d.id"))


class D(Base):
    __tablename__ = "d"

    id = mapped_column(Integer, primary_key=True)

relationship.secondary relationship.primaryjoin relationship.secondaryjoin a b c d A D I

sess.scalars(select(A).join(A.d)).all()

{execsql}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

:func:`_orm.relationship`I

Предупреждение

``viewonly=True``A

Взаимосвязь с классом Aliased¶

Добавлено в версии 1.3: AliasedClass relationship() AliasedClass T

relationship.secondary A B C D A B A B relationship.primaryjoin relationship.secondary A->secondary->B A B AliasedClass aliased() relationship() I

relationship() A B C D A B B

class A(Base):
    __tablename__ = "a"

    id = mapped_column(Integer, primary_key=True)
    b_id = mapped_column(ForeignKey("b.id"))


class B(Base):
    __tablename__ = "b"

    id = mapped_column(Integer, primary_key=True)


class C(Base):
    __tablename__ = "c"

    id = mapped_column(Integer, primary_key=True)
    a_id = mapped_column(ForeignKey("a.id"))

    some_c_value = mapped_column(String)


class D(Base):
    __tablename__ = "d"

    id = mapped_column(Integer, primary_key=True)
    c_id = mapped_column(ForeignKey("c.id"))
    b_id = mapped_column(ForeignKey("b.id"))

    some_d_value = mapped_column(String)


# 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.scalars(select(A).join(A.b)).all()

{execsql}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 в запросах¶

A.b B_viacd B A.b B_viacd B A.b I

subq = select(B).join(D, D.b_id == B.id).join(C, C.id == D.c_id).subquery()

B_viacd_subquery = aliased(B, subq)

A.b = relationship(B_viacd_subquery, primaryjoin=A.b_id == subq.c.id)

``A.b``A

sess.scalars(select(A).join(A.b)).all()

{execsql}SELECT a.id AS a_id, a.b_id AS a_b_id
FROM a JOIN (SELECT b.id AS id, b.some_b_column AS some_b_column
FROM b JOIN d ON d.b_id = b.id JOIN c ON c.id = d.c_id) AS anon_1 ON a.b_id = anon_1.id

A.b B_viacd_subquery B I

sess.scalars(
    select(A)
    .join(A.b)
    .where(B_viacd_subquery.some_b_column == "some b")
    .order_by(B_viacd_subquery.id)
).all()

{execsql}SELECT a.id AS a_id, a.b_id AS a_b_id
FROM a JOIN (SELECT b.id AS id, b.some_b_column AS some_b_column
FROM b JOIN d ON d.b_id = b.id JOIN c ON c.id = d.c_id) AS anon_1 ON a.b_id = anon_1.id
WHERE anon_1.some_b_column = ? ORDER BY anon_1.id

Отношения с ограничением по строкам с помощью оконных функций¶

:class:`.AliasedClass`A

class A(Base):
    __tablename__ = "a"

    id = mapped_column(Integer, primary_key=True)


class B(Base):
    __tablename__ = "b"
    id = mapped_column(Integer, primary_key=True)
    a_id = mapped_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() W

for a1 in session.scalars(select(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», мы получим первые десять «bs», упорядоченные по «b.id». Разбиение по «a_id» гарантирует, что каждый «номер строки» будет локальным для родительского «a_id».

Такое отображение, как правило, также включает «простое» отношение от «A» к «B» для операций сохранения, а также когда требуется полный набор объектов «B» на «A».

Построение свойств с поддержкой запросов¶

relationship.viewonly relationship() Query V

class User(Base):
    __tablename__ = "user"
    id = mapped_column(Integer, primary_key=True)

    @property
    def addresses(self):
        return object_session(self).query(Address).with_parent(self).filter(...).all()

:ref:`mapper_hybrids`I

См.также

Использование дескрипторов и гибридов

Отношения в списке смежности

Работа с большими коллекциями

Вернуться на верх

Настройка способа присоединения отношений¶

Работа с несколькими путями присоединения¶

Указание альтернативных условий присоединения¶

Создание пользовательских иностранных условий¶

Использование пользовательских операторов в условиях присоединения¶

Пользовательские операторы на основе функций SQL¶

Перекрывающиеся внешние ключи¶

Нереляционные сравнения / Материализованный путь¶

Отношения «многие-ко-многим» с самореференцией¶

Составные «вторичные» соединения¶

Взаимосвязь с классом Aliased¶

Использование цели AliasedClass в запросах¶

Отношения с ограничением по строкам с помощью оконных функций¶

Построение свойств с поддержкой запросов¶

SQLAlchemy 2.0

Содержание

Дополнительно

Вы здесь: