Configuring how Relationship Joins¶

relationship() は通常、2つのテーブル間の外部キー関係を調べてどの列を比較すべきかを決定することで、2つのテーブル間の結合を作成します。この動作をカスタマイズする必要がある状況はさまざまです。

Handling Multiple Join Paths¶

対処すべき最も一般的な状況の1つは、2つのテーブル間に複数の外部キーパスがある場合です。

Address クラスへの2つの外部キーを含む Customer クラスを考えてみましょう。:

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.

上記のメッセージはかなり長いです。 relationship() が返す可能性のあるメッセージはたくさんありますが、それらはさまざまな一般的な設定の問題を検出するように注意深く調整されています。そのほとんどは、あいまいさやその他の欠落した情報を解決するために必要な追加の設定を示唆しています。

この場合、メッセージはそれぞれの relationship() にどの外部キー列を考慮すべきかを指示することで、それぞれの relationship() を修飾することを要求しています。適切な形式は次のとおりです。:

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 オブジェクトから Customer.billing_address 関係をロードすると、ロードされる Address の行を識別するために billing_address_id に存在する値が使用されます。同様に、 shipping_address 関係には shipping_address_id が使用されます。2つの列のリンクも持続性の際に役割を果たします。挿入されたばかりの Address オブジェクトの新しく生成された主キーは、フラッシュ時に関連する Customer オブジェクトの適切な外部キー列にコピーされます。

Declarativeで foreign_keys を指定する場合、文字列名を使用して指定することもできますが、リストを使用する場合は、 リストが文字列 の一部であることが重要です:

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

この例では、必要な Column が1つしかないので、このリストは必要ありません:

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

Warning

Pythonで評価可能な文字列として渡された場合、 relationship.foreign_keys 引数はPythonの eval() 関数を使って解釈されます。 この文字列に信頼できない入力を渡してはいけません 。relationship() 引数の宣言的な評価の詳細については Evaluation of relationship arguments を参照してください。

Specifying Alternate Join Conditions¶

ジョインを作成する際の relationship() のデフォルトの動作では、一方の側のプライマリキー列の値が、他方の側の外部キー参照列の値と等しくなります。この基準は、 relationship.primaryjoin 引数を使用して任意に変更できます。また、「セカンダリ」テーブルが使用されている場合は、 relationship.secondaryjoin 引数を使用して変更できます。

以下の例では、 User クラスと番地を格納する Address クラスを使って、 Boston という都市を指定する Address オブジェクトだけをロードする関係 boston_addresses を作成します。:

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)

この文字列SQL式の中で、 and_() 結合構文を使って、結合条件の2つの異なる述語を確立しました。つまり、 User.id 列と Address.user_id 列の両方を互いに結合し、 Address の行を city='Boston' だけに制限しました。宣言型を使用すると、 and_() のような初歩的なSQL関数が、文字列 relationship() 引数の評価された名前空間で自動的に使用可能になります。

Warning

Pythonで評価可能な文字列として渡された場合、 relationship.primaryjoin `引数はPythonの ``eval()` 関数を使って解釈されます。 この文字列に信頼できない入力を渡してはいけません 。 relationship() 引数の宣言的な評価の詳細については Evaluation of relationship arguments を参照してください。

relationship. primaryjoin で使用するカスタム基準は、一般に、この関係をロードまたは表現するためにSQLAlchemyがSQLをレンダリングしている場合にのみ有効です。つまり、属性ごとの遅延ロードを実行するために発行されるSQL文で使用されるか、 Select.join() や熱心な”joined”または”subquery”スタイルのロードなどによってクエリ時に結合が構築されるときに使用されます。メモリ内のオブジェクトが操作されているとき、私たちは、私たちが望む任意の Address オブジェクトを、 boston_addresses コレクションに入れることができます。これは、 .city 属性の値が何であるかに関係ありません。オブジェクトは、属性が期限切れになり、基準が適用されるデータベースから再ロードされるまで、コレクション内に存在します。フラッシュが発生すると、 boston_addresses 内のオブジェクトは無条件にフラッシュされ、主キーの user.id 列の値が、各行の外部キーを保持する address.user_id 列に割り当てられます。 city 基準は、ここでは効果がありません。なぜなら、フラッシュプロセスは、主キー値を参照外部キー値に同期させることだけを気にするからです。

Creating Custom Foreign Conditions¶

プライマリ結合条件のもう1つの要素は、「外部」と見なされる列がどのように決定されるかです。通常、 Column オブジェクトのサブセットは、 ForeignKey を指定するか、結合条件に関連する ForeignKeyConstraint の一部になります。 relationship() は、この外部キーの状態を見て、この関係のデータをどのようにロードして保持するかを決定します。ただし、 relationship. primaryjoin 引数を使用して、”スキーマ”レベルの外部キーを含まない結合条件を作成できます。このような結合を確立するために、 relationship. primaryjoin を relationship. foreign_keys および relationship. remote_side と明示的に組み合わせることができます。

以下では、クラス HostEntry がそれ自身に結合され、文字列 content 列が ip_address 列に等しくなります。これは INET と呼ばれるPostgreSQLの型です。結合の一方の側を他方の型にキャストするために、 cast() を使用する必要があります:

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)

上記の別の構文として、 relationship.primaryjoin 式の中でインラインで foreign() および remote()annotations を使用することもできます。この構文は、 relationship.foreign_keys および relationship.remote_side 引数が与えられた場合に、 relationship() が通常それ自身で結合条件に適用する注釈を表します。これらの関数は、明示的な結合条件が存在する場合にはより簡潔になり、さらに、その列が複数回記述されているか、複雑なSQL式の中にあるかにかかわらず、”foreign”または”remote”である列を正確にマークするのに役立ちます。:

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),
    )

Using custom operators in join conditions¶

関係のもう1つのユースケースは、 _PostgreSQL.INET や _PostgreSQL.CIDR のような型と結合するときに、PostgreSQLの << 演算子のようなカスタム演算子を使用することです。カスタムブール演算子には Operators.bool_op() 関数を使用します:

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

上記のような比較は、 relationship() を構築する際に、 relationship.primaryjoin と直接使用することができます:

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

Custom operators based on SQL functions¶

Operators.op.is_comparison のユースケースの変形として、演算子ではなくSQL関数を使用する場合があります。このユースケースの典型的な例はPostgreSQL PostGIS関数ですが、バイナリ条件を解決する任意のデータベース上の任意のSQL関数が適用される可能性があります。このユースケースに適合するように、 FunctionElement.as_comparison() メソッドは、 func 名前空間から呼び出されるような任意のSQL関数を変更して、関数が2つの式の比較を生成することをORMに示すことができます。次の例は、これを Geoalchemy2 ライブラリで説明しています:

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() は、SQL関数の func.ST_Contains() が Polygon.geom 式と Point.geom 式を比較していることを示しています。 foreign() アノテーションはさらに、この特定の関係においてどの列が”外部キー”の役割を持つかを示しています。

New in version 1.3: Added FunctionElement.as_comparison().

Overlapping Foreign Keys¶

複合外部キーを使用すると、1つのカラムが外部キー制約によって参照される複数のカラムの対象になるなど、まれなシナリオが発生することがあります。

上記のマッピングを設定すると、次の警告が表示されます。:

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 が2つの異なる外部キー制約の対象であるという事実に由来しています。 Magazine.id を直接ソース列として参照しますが、 Writer.magazine_id も Writer への複合キーのコンテキストでソース列として参照します。 Article を特定の Magazine に関連付け、その Article を*別の* Magazine に関連付けられた Writer に関連付けると、ORMは Article.magazine_id を非確定的に上書きし、参照する雑誌を静かに変更します。また、 Writer を Article から関連付けを解除すると、この列にNULLを配置しようとすることもあります。この警告は、これが事実であることを示しています。

To solve this, we need to break out the behavior of Article to include all three of the following features:

これを解決するには、 Article の振る舞いを分析して、以下の3つの特徴をすべて含める必要があります。

Article は、何よりもまず、 Article.magazine 関係のみに存在するデータに基づいて Article.magazine_id に書き込みます。これは Magazine.id からコピーされた値です。
Article can write to Article.writer_id on behalf of data persisted in the Article.writer relationship, but only the Writer.id column; the Writer.magazine_id column should not be written into Article.magazine_id as it ultimately is sourced from Magazine.id.

Article は Article.writer の関係にあるデータの代わりに Article.writer_id に書き込むことができますが、それは Writer.id 列だけです。 Writer.magazine_id 列は、最終的には Magazine.id から供給されるので、 Article.magazine_id に書き込むべきではありません。
Article takes Article.magazine_id into account when loading Article.writer, even though it doesn’t write to it on behalf of this relationship.

Article は Article.writer をロードする時に Article.magazine_id を考慮に入れますが、この関係のために書き込みは しません 。

To get just #1 and #2, we could specify only Article.writer_id as the “foreign keys” for Article.writer:

#1と#2だけを取得するには、 ``Article.writer`` の ``外部キー`` として ``Article.writer_id`` だけを指定します。

class Article(Base):
# …

writer = relationship(“Writer”, foreign_keys=”Article.writer_id”)

However, this has the effect of Article.writer not taking Article.magazine_id into account when querying against Writer:

しかし、これには、 Writer に対して問い合わせるときに Article.magazine_id を考慮しない Article.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のすべてを取得するには、 relationship.primaryjoin を relationship.foreign_keys 引数とともに完全に組み合わせるか、より簡潔に foreign() で注釈を付けることによって、結合条件とどの列を書き込むかを表現します:

class Article(Base):
    # ...

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

Non-relational Comparisons / Materialized Path¶

Warning

this section details an experimental feature.

カスタム式を使用することは、通常のプライマリ/外部キーモデルに従わない非オーソドックスな結合条件を生成できることを意味する。そのような例の1つは、実体化されたパスパターンであり、ツリー構造を生成するために、重複するパストークンの文字列を比較する。

foreign() と remote() を慎重に使用することで、初歩的な実体化パスシステムを効果的に生成する関係を構築することができます。基本的に、 foreign() と remote() が比較式の*同じ*側にある場合、関係は「1対多」と見なされます。それらが*異なる*側にある場合、関係は「多対1」と見なされます。ここで使用する比較では、コレクションを処理するので、「1対多」として設定されたものを保持します:

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,
    )

上記では、path属性が /foo/bar2 の Element オブジェクトが与えられた場合、以下のような Element.descendants のロードを求めます。

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

Self-Referential Many-to-Many Relationship¶

See also

この節では、 Adjacency List Relationships で説明されている”adjacency list”パターンの2つのテーブルからなる変形について説明します。 Self-Referential Query Strategies と Configuring Self-Referential Eager Loading のサブセクションで、ここで説明するマッピングパターンにも同様に適用できる、自己参照の問い合わせパターンを確認してください。

多対多の関係は、 relationship.primaryjoin と relationship.secondaryjoin のどちらかまたは両方によってカスタマイズできます。後者は、 relationship.secondary 引数を使用して多対多の参照を指定する関係にとって重要です。 relationship.primaryjoin と relationship.secondaryjoin の使用を含む一般的な状況は、以下に示すように、クラスからそれ自身への多対多の関係を確立する場合です。:

from typing import List

from sqlalchemy import Integer, ForeignKey, Column, Table
from sqlalchemy.orm import DeclarativeBase, Mapped
from sqlalchemy.orm import mapped_column, 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[int] = mapped_column(primary_key=True)
    label: Mapped[str]
    right_nodes: Mapped[List["Node"]] = relationship(
        "Node",
        secondary=node_to_node,
        primaryjoin=id == node_to_node.c.left_node_id,
        secondaryjoin=id == node_to_node.c.right_node_id,
        back_populates="left_nodes",
    )
    left_nodes: Mapped[List["Node"]] = relationship(
        "Node",
        secondary=node_to_node,
        primaryjoin=id == node_to_node.c.right_node_id,
        secondaryjoin=id == node_to_node.c.left_node_id,
        back_populates="right_nodes",
    )

上記の場合、SQLAlchemyは、どの列が right_nodes と left_nodes のどちらの関係に接続すべきかを自動的に知ることはできません。 relationship.primaryjoin 引数と relationship.secondaryjoin 引数は、関連付けテーブルへの結合方法を設定します。上記の宣言形式では、これらの条件を「Node」クラスに対応するPythonブロック内で宣言しているので、「id」変数は結合したい Column オブジェクトとして直接利用できます。

別の方法として、文字列を使って relationship.primaryjoin 引数と relationship.secondaryjoin 引数を定義することもできます。これは、設定にまだ利用可能な Node.id 列オブジェクトがない場合や、おそらくまだ node_to_node テーブルが利用できない場合に適しています。宣言文字列内のプレーンな Table オブジェクトを参照するときは、 MetaData 内に存在するテーブルの文字列名を使用します:

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",
    )

Warning

Pythonで評価可能な文字列として渡された場合、 relationship.primaryjoin 引数と relationship.secondaryjoin 引数はPythonの eval() 関数を使って解釈されます。 これらの文字列に信頼できない入力を渡してはいけません 。 relationship() 引数の宣言的な評価の詳細については Evaluation of relationship arguments を参照してください。

ここでの古典的なマッピングの状況も同様で、 node_to_node を node.c.id に結合することができます。:

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 backrefを指定していることに注意してください。 relationship() が2番目の関係を逆方向に作成する場合、 relationship.primaryjoin 引数と relationship.secondaryjoin 引数を逆にするのが賢明です。

See also

Adjacency List Relationships - 単一テーブルのバージョン
Self-Referential Query Strategies - 自己参照マッピングを使った問い合わせのヒント
Configuring Self-Referential Eager Loading - 自己参照マッピングによるeager loadingのヒント

Composite “Secondary” Joins¶

Note

このセクションでは、SQLAlchemyである程度サポートされている遠端のケースを取り上げますが、このような問題は、合理的なリレーショナルレイアウトや in-Python attributes を使用して、可能な限り簡単な方法で解決することをお勧めします。

2つのテーブル間に relationship() を構築しようとするとき、それらを結合するために2つまたは3つ以上のテーブルが必要になることがあります。これは relationship() の領域であり、何が可能かの境界を押し広げようとするものであり、多くの場合、これらのエキゾチックなユースケースの多くに対する最終的な解決策は、SQLAlchemyメーリングリストで打ち出される必要があります。

最近のバージョンのSQLAlchemyでは、 relationship.secondary パラメータを使用して、複数のテーブルで構成される複合ターゲットを提供する場合があります。このような結合条件の例を次に示します(現状のまま機能するには、少なくともバージョン0.9.2が必要です):

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)

上の例では、以下の3つ全てを提供しています relationship.secondary,:paramref:_orm.relationship.primaryjoin,:paramref:_orm.relationship.secondaryjoin 名前付きのテーブル a , b , c , d を直接参照する宣言形式です。 A から D への問い合わせは以下のようになります:

sess.scalars(select(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つの”テーブルがあるだけで、複雑さは中間に保たれます。

Warning

上記のような関係は通常、 relationship.viewonly を使用して viewonly=True とマークされ、読み取り専用と見なされます。上記のような関係を書き込み可能にする方法は時々ありますが、これは一般的に複雑でエラーが発生しやすくなります。

Relationship to Aliased Class¶

前節では、結合条件内に追加のテーブルを配置するために relationship. secondary を使用するテクニックを説明しました。このテクニックでも十分でない複雑な結合のケースが1つあります。 A から B に結合しようとするとき、その間に任意の数の C 、 D などを使用しますが、 A と B の間にも*直接*結合条件があります。この場合、 A から B への結合は、複雑な relationship. primaryjoin 条件だけでは表現が困難な場合があります。これは、中間のテーブルが特別な処理を必要とする可能性があるためです。また、 A->secondary->B パターンは A と B の間の参照を直接サポートしていないため、 relationship. secondary オブジェクトで表現することもできません。この**非常に高度な**ケースが発生した場合は、関係のターゲットとして2番目のマッピングを作成することができます。ここでは、この結合に必要なすべての追加テーブルを含むクラスへのマッピングを作成するために AliasedClass を使用します。このマッパーをクラスの 代替 マッピングとして生成するために、 aliased() 関数を使用して新しい構成体を生成し、オブジェクトに対して relationship() を単純なマップクラスであるかのように使用します。

以下は relationship() の単純な A から B への結合を示していますが、プライマリ結合の条件は2つの追加のエンティティ C と D で補強されています。これらのエンティティは同時に A と 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()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

Integrating AliasedClass Mappings with Typing and Avoiding Early Mapper Configuration¶

マップされたクラスに対して aliased() 構文を作成すると、 configure_mappers() ステップが強制的に実行され、現在のすべてのクラスとその関係が解決されます。これは、現在のマッピングで必要とされる無関係なマップされたクラスがまだ宣言されていない場合や、関係の設定自体がまだ宣言されていないクラスにアクセスする必要がある場合に問題になる可能性があります。さらに、SQLAlchemyの宣言型パターンは、関係が事前に宣言されている場合にPythonの型付けと最も効果的に連携します。

これらの問題に対処するための関係の構築を組織化するために、 MapperEvents.before_mapper_configured() のような構成レベルのイベントフックを使用することができます。これは、すべてのマッピングが構成の準備ができたときにのみ構成コードを呼び出します:

from sqlalchemy import event

class A(Base):
    __tablename__ = "a"

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

@event.listens_for(A, "before_mapper_configured")
def _configure_ab_relationship(mapper, cls):
    # do the above configuration in a configuration hook

    j = join(B, D, D.b_id == B.id).join(C, C.id == D.c_id)
    B_viacd = aliased(B, j, flat=True)
    A.b = relationship(B_viacd, primaryjoin=A.b_id == j.c.b_id)

上記では、完全に設定されたバージョンの A が要求された場合にのみ関数 _configure_ab_relationship() が呼び出され、その時点でクラス B 、 D 、 C が使用可能になります。

インライン型付けと統合するアプローチでは、同様のテクニックを使用して、エイリアスされたクラスの”シングルトン”作成パターンを効果的に生成することができます。エイリアスされたクラスは、グローバル変数として後で初期化され、インラインの関係で使用できます:

from typing import Any

B_viacd: Any = None
b_viacd_join: Any = None

class A(Base):
    __tablename__ = "a"

    id: Mapped[int] = mapped_column(primary_key=True)
    b_id: Mapped[int] = mapped_column(ForeignKey("b.id"))

    # 1. the relationship can be declared using lambdas, allowing it to resolve
    #    to targets that are late-configured
    b: Mapped[B] = relationship(
        lambda: B_viacd, primaryjoin=lambda: A.b_id == b_viacd_join.c.b_id
    )

# 2. configure the targets of the relationship using a before_mapper_configured
#    hook.
@event.listens_for(A, "before_mapper_configured")
def _configure_ab_relationship(mapper, cls):
    # 3. set up the join() and AliasedClass as globals from within
    #    the configuration hook.

    global B_viacd, b_viacd_join

    b_viacd_join = join(B, D, D.b_id == B.id).join(C, C.id == D.c_id)
    B_viacd = aliased(B, b_viacd_join, flat=True)

Using the AliasedClass target in Queries¶

前の例では、 A.b 関係はターゲットとして B_viacd エンティティを参照しており、 B クラスを直接 参照していません 。 A.b 関係を含む追加の基準を追加するには、特に A.b のターゲットエンティティがエイリアスまたはサブクエリに変換される場合には、 B を使用するのではなく、 B_viacd を直接参照する必要があります。以下に、結合ではなくサブクエリを使用した同じ関係を示します。

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」の関係を使った問い合わせは副問い合わせになります。:

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

sess.scalars(select(A).join(A.b).where(B_viacd_subquery.some_b_column=="some b").order_by(B_viacd_subquery.id)).all()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

Row-Limited Relationships with Window Functions¶

AliasedClass オブジェクトへの関係のもう1つの興味深いユースケースは、関係が任意の形式の特殊なSELECTに結合する必要がある場合です。1つのシナリオは、関係に対して返される行数を制限するなど、Window関数の使用が必要な場合です。次の例は、各コレクションの最初の10項目をロードする非プライマリマッパー関係を示しています。:

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)
)

selectinload() のようなほとんどのローダ戦略では、上記の partitioned_bs の関係を使うことができます:

for a1 in session.scalars(select(A).options(selectinload(A.partitioned_bs))):
    print(a1.partitioned_bs)  # <-- will be no more than ten objects

Where above, the “selectinload” query looks like:

上記の場合、「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”で順序付けられた最初の10個の”bs”が取得されます。”a_id”でパーティション化することにより、各”行番号”が親”a_id”に対してローカルであることが保証されます。

Such a mapping would ordinarily also include a “plain” relationship from “A” to “B”, for persistence operations as well as when the full set of “B” objects per “A” is desired.

このようなマッピングは、通常、”A”ごとの”B”オブジェクトの完全なセットが必要な場合だけでなく、パーシスタンス操作のために、”A”から”B”への”単純な”関係も含みます。

Building Query-Enabled Properties¶

非常に野心的なカスタム結合条件は、直接永続化できない可能性があり、場合によっては正しくロードされないことさえあります。式の永続化部分を削除するには、フラグ relationship.viewonly を relationship() で使用します。これにより、読み取り専用属性として確立されます(コレクションに書き込まれたデータはflush()で無視されます)。ただし、極端な場合には、次のように Query と組み合わせて通常のPythonプロパティを使用することを検討してください。

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

その他のケースでは、既存のPython内のデータを利用するように記述子を構築することができます。特殊なPython属性のより一般的な議論については、 Using Descriptors and Hybrids のセクションを参照してください。

See also

Using Descriptors and Hybrids

Notes on using the viewonly relationship parameter¶

relationship.viewonly パラメータを relationship() 構文に適用すると、この relationship() がORM unit of work 操作に関与しないことを示します。さらに、この属性は、表現されたコレクションのPython内のミューテーションに関与することを想定していません。つまり、viewonly関係はリストやセットのような変更可能なPythonコレクションを参照することができますが、そのリストやセットをマップされたインスタンスに存在するように変更しても、ORMフラッシュプロセスには**何の影響も**ありません。

このシナリオを検討するには、次のマッピングを検討します。:

from __future__ import annotations

import datetime

from sqlalchemy import and_
from sqlalchemy import ForeignKey
from sqlalchemy import func
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 | None]

    all_tasks: Mapped[list[Task]] = relationship()

    current_week_tasks: Mapped[list[Task]] = relationship(
        primaryjoin=lambda: and_(
            User.id == Task.user_account_id,
            # this expression works on PostgreSQL but may not be supported
            # by other database engines
            Task.task_date >= func.now() - datetime.timedelta(days=7),
        ),
        viewonly=True,
    )

class Task(Base):
    __tablename__ = "task"

    id: Mapped[int] = mapped_column(primary_key=True)
    user_account_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
    description: Mapped[str | None]
    task_date: Mapped[datetime.datetime] = mapped_column(server_default=func.now())

    user: Mapped[User] = relationship(back_populates="current_week_tasks")

次のセクションでは、この設定のさまざまな側面について説明します。

In-Python mutations including backrefs are not appropriate with viewonly=True¶

上記のマッピングは、 User.current_week_tasks のviewonly関係を、 Task.user 属性の backref ターゲットとしてターゲットにしています。これは現在、SQLAlchemyのORM設定プロセスではフラグ付けされていませんが、設定エラーです。 Task の .user 属性を変更しても、 .current_week_tasks 属性には影響しません。

>>> u1 = User()
>>> t1 = Task(task_date=datetime.datetime.now())
>>> t1.user = u1
>>> u1.current_week_tasks
[]

relationship.sync_backrefs という別のパラメータがあります。これはここで有効にして、この場合に .current_week_tasks を変更できるようにすることができます。しかし、これはビューのみの関係でのベストプラクティスとは考えられておらず、代わりにPython内の変更に頼るべきではありません。

このマッピングでは、 User.all_tasks と Task.user はどちらも表示専用ではなく、正常に同期するので、これらの間にバックレフを設定できます。

Beyond the issue of backref mutations being disabled for viewonly relationships, plain changes to the User.all_tasks collection in Python are also not reflected in the User.current_week_tasks collection until changes have been flushed to the database.

表示のみの関係に対してbackref mutationが無効になる問題以外にも、Pythonでの User.all_tasks コレクションへの単純な変更は、変更がデータベースにフラッシュされるまで User.current_week_tasks コレクションに反映されません。

全体として、カスタムコレクションがPython内のミューテーションに即座に応答する必要があるユースケースでは、ビューのみの関係は一般に適切ではありません。より良いアプローチは、SQLAlchemyの Hybrid Attributes 機能を使用することです。または、インスタンスのみのケースでは、現在のPythonインスタンスに関して生成されたユーザ定義のコレクションを実装できるPythonの @property を使用します。このように動作するように例を変更するには、 Task.user の relationship.back_populates パラメータを修復して User.all_tasks を参照し、直接の User.all_tasks コレクションに関して結果を提供する単純な @property を示します:

class User(Base):
    __tablename__ = "user_account"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str | None]

    all_tasks: Mapped[list[Task]] = relationship(back_populates="user")

    @property
    def current_week_tasks(self) -> list[Task]:
        past_seven_days = datetime.datetime.now() - datetime.timedelta(days=7)
        return [t for t in self.all_tasks if t.task_date >= past_seven_days]

class Task(Base):
    __tablename__ = "task"

    id: Mapped[int] = mapped_column(primary_key=True)
    user_account_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
    description: Mapped[str | None]
    task_date: Mapped[datetime.datetime] = mapped_column(server_default=func.now())

    user: Mapped[User] = relationship(back_populates="all_tasks")

毎回オンザフライで計算されるPython内のコレクションを使用すると、データベースをまったく使用せずに、常に正しい答えが得られることが保証されます:

>>> u1 = User()
>>> t1 = Task(task_date=datetime.datetime.now())
>>> t1.user = u1
>>> u1.current_week_tasks
[<__main__.Task object at 0x7f3d699523c0>]

viewonly=True collections / attributes do not get re-queried until expired¶

元のviewonly属性を続けて、実際に persistent オブジェクトの User.all_tasks コレクションを変更した場合、viewonlyコレクションは、 2つのことが起こった後に、この変更の最終的な結果を表示することができます。1つ目は、 User.all_tasks への変更が flushed され、少なくともローカルトランザクションのスコープ内で新しいデータがデータベースで利用できるようになります。2つ目は、 User.current_week_tasks 属性が expired になり、新しいSQLクエリを介してデータベースに再ロードされます。

この要件をサポートするために使用する最も単純なフローは、 viewonly関係が、主にで始まる読み取り専用の操作でのみ使用される フローです。以下のように、データベースから User を新たに取得すると、そのコレクションは現在のものになります。:

>>> with Session(e) as sess:
...     u1 = sess.scalar(select(User).where(User.id == 1))
...     print(u1.current_week_tasks)
[<__main__.Task object at 0x7f8711b906b0>]

上の例では、 Session.commit() の呼び出しによって、 u1.all_tasks への変更がデータベースにフラッシュされ、すべてのオブジェクトが期限切れになりました。そのため、 u1.current_week_tasks にアクセスすると、 lazy load が発生し、この属性の内容がデータベースから新たに取り出されました。

実際にトランザクションをコミットせずに操作をインターセプトするには、属性を最初に明示的に expired にする必要があります。これを行う簡単な方法は、単に直接呼び出すことです。次の例では、 Session.flush() が保留中の変更をデータベースに送信し、次に Session.expire() を使用して u1.current_week_tasks コレクションを期限切れにして、次のアクセス時に再フェッチされるようにします。

>>> with Session(e) as sess:
...     u1 = sess.scalar(select(User).where(User.id == 1))
...     u1.all_tasks.append(Task(task_date=datetime.datetime.now()))
...     sess.flush()
...     sess.expire(u1, ["current_week_tasks"])
...     print(u1.current_week_tasks)
[<__main__.Task object at 0x7fd95a4c8c50>, <__main__.Task object at 0x7fd95a4c8c80>]

Session.autoflush をデフォルト値の True に保つ Session を仮定すると、実際には Session.flush() の呼び出しをスキップすることができます。 current_week_tasks 属性は、有効期限が切れた後にアクセスされるとautoflushをトリガするからです:

>>> with Session(e) as sess:
...     u1 = sess.scalar(select(User).where(User.id == 1))
...     u1.all_tasks.append(Task(task_date=datetime.datetime.now()))
...     sess.expire(u1, ["current_week_tasks"])
...     print(u1.current_week_tasks)  # triggers autoflush before querying
[<__main__.Task object at 0x7fd95a4c8c50>, <__main__.Task object at 0x7fd95a4c8c80>]

さらに手の込んだものへの上記のアプローチを続けると、 event hooks を使用して、関連する`User.all_tasks`コレクションが変更されたときに、プログラムで有効期限を適用することができます。これは**高度なテクニック**であり、そこでは、 @property のようなより単純なアーキテクチャや、読み取り専用のユースケースに固執することが最初に検討されるべきです。私たちの単純な例では、これは次のように設定されます:

from sqlalchemy import event, inspect

@event.listens_for(User.all_tasks, "append")
@event.listens_for(User.all_tasks, "remove")
@event.listens_for(User.all_tasks, "bulk_replace")
def _expire_User_current_week_tasks(target, value, initiator):
    inspect(target).session.expire(target, ["current_week_tasks"])

上記のフックでは、変更操作がインターセプトされ、自動的に User.current_week_tasks コレクションの有効期限が切れます。:

>>> with Session(e) as sess:
...     u1 = sess.scalar(select(User).where(User.id == 1))
...     u1.all_tasks.append(Task(task_date=datetime.datetime.now()))
...     print(u1.current_week_tasks)
[<__main__.Task object at 0x7f66d093ccb0>, <__main__.Task object at 0x7f66d093cce0>]

上で使用した AttributeEvents イベントフックもbackref変換によってトリガされますので、上のフックを使って Task.user への変更もインターセプトされます:

>>> with Session(e) as sess:
...     u1 = sess.scalar(select(User).where(User.id == 1))
...     t1 = Task(task_date=datetime.datetime.now())
...     t1.user = u1
...     sess.add(t1)
...     print(u1.current_week_tasks)
[<__main__.Task object at 0x7f3b0c070d10>, <__main__.Task object at 0x7f3b0c057d10>]

SQLAlchemy 2.0 Documentation - 私家版日本語訳ドキュメント

SQLAlchemy 2.0 Documentation - 私家版日本語訳ドキュメント

SQLAlchemy ORM

Configuring how Relationship Joins¶

Handling Multiple Join Paths¶

Specifying Alternate Join Conditions¶

Creating Custom Foreign Conditions¶

Using custom operators in join conditions¶

Custom operators based on SQL functions¶

Overlapping Foreign Keys¶

Non-relational Comparisons / Materialized Path¶

Self-Referential Many-to-Many Relationship¶

Composite “Secondary” Joins¶

Relationship to Aliased Class¶

Integrating AliasedClass Mappings with Typing and Avoiding Early Mapper Configuration¶

Using the AliasedClass target in Queries¶

Row-Limited Relationships with Window Functions¶

Building Query-Enabled Properties¶

Notes on using the viewonly relationship parameter¶

In-Python mutations including backrefs are not appropriate with viewonly=True¶

viewonly=True collections / attributes do not get re-queried until expired¶