SQLAlchemy ORMの宣言的モデル定義でMany to Manyのrelationshipがある際にモデルファイルを複数に分ける

Python

SQLAlchemy

Kazuki Moriyama (森山和樹)

この記事で書くこと

SQLAlchemyで宣言的モデルを複数のファイルで定義する際に生じる

型エラー
実行時エラー

の解決について書く。

宣言的モデルとは

ORMのモデルをPythonのクラスを用いて宣言的(Declarative)に書ける機能。

from sqlalchemy import Integer, String, ForeignKey
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column

# declarative base class
class Base(DeclarativeBase):
    pass

# an example mapping using the base
class User(Base):
    __tablename__ = "user"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    fullname: Mapped[str] = mapped_column(String(30))
    nickname: Mapped[Optional[str]]

この様に書くことでDBのテーブルの構造、及びテーブルにマッピングされるクラスの構造を見たまま定義できる。

これとの対比で命令的(Imperative)にモデルを定義する方法もあるが、そちらではテーブルの構造とPythonクラスへのマッピング設定をsqlalchemyのメソッド呼び出し（命令)によって定義する。

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

mapper_registry = registry()

user_table = Table(
    "user",
    mapper_registry.metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String(50)),
    Column("fullname", String(50)),
    Column("nickname", String(12)),
)

class User:
    pass

mapper_registry.map_imperatively(User, user_table)

宣言的モデルによるMany to Manyの定義

いわゆる中間テーブルを利用した多対多のモデル定義は以下のようになる。

from typing import Optional

from sqlalchemy.schema import ForeignKey
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import relationship

class Base(DeclarativeBase):
    pass

class Association(Base):
    __tablename__ = "association_table"

    left_id: Mapped[int] = mapped_column(ForeignKey("left_table.id"), primary_key=True)
    right_id: Mapped[int] = mapped_column(
        ForeignKey("right_table.id"), primary_key=True
    )
    extra_data: Mapped[Optional[str]]

    child: Mapped["Child"] = relationship(back_populates="parent_associations")

    parent: Mapped["Parent"] = relationship(back_populates="child_associations")

class Parent(Base):
    __tablename__ = "left_table"

    id: Mapped[int] = mapped_column(primary_key=True)

    children: Mapped[list["Child"]] = relationship(
        secondary="association_table", back_populates="parents"
    )

    child_associations: Mapped[list["Association"]] = relationship(
        back_populates="parent"
    )

class Child(Base):
    __tablename__ = "right_table"

    id: Mapped[int] = mapped_column(primary_key=True)

    parents: Mapped[list["Parent"]] = relationship(
        secondary="association_table", back_populates="children"
    )

    parent_associations: Mapped[list["Association"]] = relationship(
        back_populates="child"
    )

このモデル定義ではテーブルのカラム定義としてidやそれぞれ外部キー定義がされていると共に、relationshipによって参照先のモデルがフィールドとして定義されている。
これによってアプリケーション上で参照先のモデルが簡易にDBからロード・取得できるようになっている。

parent: Parent = ...

parent.children # Childモデルがロードされる

ここまでがよくあるsqlalchemyのモデル定義でこれを1ファイルで定義している際は問題ないが、複数ファイルに分けて定義すると問題が起きる。
そしてテーブルが増えると1ファイルで定義するのは大変なのでこれらは分割したくなる。

複数ファイルにモデル定義を分ける際の問題と解決方法

最初のファイル分け

ナイーブにファイル分割すると以下のようになる。

base.py

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

parent.py

import sqlalchemy

from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

from .base import Base
from .child import Child
from .association import Association

class Parent(Base):
    __tablename__ = "left_table"

    id: Mapped[int] = mapped_column(primary_key=True)

    children: Mapped[list["Child"]] = relationship(
        secondary="association_table", back_populates="parents"
    )

    child_associations: Mapped[list["Association"]] = relationship(
        back_populates="parent"
    )

child.py

from .base import Base

from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

from .association import Association
from .parent import Parent

class Child(Base):
    __tablename__ = "right_table"

    id: Mapped[int] = mapped_column(primary_key=True)

    parents: Mapped[list["Parent"]] = relationship(
        secondary="association_table", back_populates="children"
    )

    parent_associations: Mapped[list["Association"]] = relationship(
        back_populates="child"
    )

association.py

from .base import Base

from sqlalchemy.schema import ForeignKey
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

from .child import Child
from .parent import Parent

class Association(Base):
    __tablename__ = "association_table"

    left_id: Mapped[int] = mapped_column(ForeignKey("left_table.id"), primary_key=True)
    right_id: Mapped[int] = mapped_column(
        ForeignKey("right_table.id"), primary_key=True
    )
    extra_data: Mapped[Optional[str]]

    child: Mapped["Child"] = relationship(back_populates="parent_associations")

    parent: Mapped["Parent"] = relationship(back_populates="child_associations")

これでは問題があるので一つずつ解消していく。

問題1: 循環importによる実行時エラー

現在はparent、childとassociationがお互いにimportしてimportが循環しているため、実行時にエラーがでる。

ImportError: cannot import name 'Child' from partially initialized module 'child' (most likely due to a circular import)

しかしお互いにクラスをimportしないとMapped[list["Parent"]]の様に型がつけられない。
このような場合にTYPE_CHECKINGと呼ばれる変数が用意されていて、それを使用すれば実行時ではなく型チェック時だけimportをすることによって実行時循環importを防ぐ事ができる。

parent.py

from typing import TYPE_CHECKING
from .base import Base

from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

if TYPE_CHECKING: # 型チェック時だけimportする
    from .association import Association
    from .parent import Parent

class Child(Base):
    __tablename__ = "right_table"

    id: Mapped[int] = mapped_column(primary_key=True)

    parents: Mapped[list["Parent"]] = relationship(
        secondary="association_table", back_populates="children"
    )

    parent_associations: Mapped[list["Association"]] = relationship(
        back_populates="child"
    )

他のchild.py、association.pyに対しても同じ様にimport部分を修正すればこの問題が解決できる。

問題2: モデル実行時のrelation先クラス解決

型チェック時のみ型をimportすることによって循環import問題を解決したが、実はrelationshipをうまく動かすためには実行時にも関連先のクラス情報が必要となる。
実際、循環importを解決した状態でモデル定義を使用すると以下のような実行時エラーに遭遇する。

NameError: name 'association_table' is not defined

The above exception was the direct cause of the following exception:

...

sqlalchemy.exc.InvalidRequestError: When initializing mapper Mapper[Parent(left_table)], expression 'association_table' failed to locate a name ("name 'association_table' is not defined"). If this is a class name, consider adding this relationship() to the <class 'separate.parent.Parent'> class after both dependent classes have been defined.

relationshipは実行時に関連先のデータをロードするために関連先のテーブルを表すクラスの情報がほしいのだが、そのクラス情報がimportなどによってスコープに入っていないとエラーが出る。
同じファイルにすべてのモデルを定義している際にはすべて同じスコープにクラスが入っているので、特殊なことをせずともrelationshipは関連先のクラスをMappedの型情報から取得できる。
循環importの解決のためにimportを型チェック時だけ行い、実行時には関連先クラスの情報がimportされていないのがこのエラーの原因。

解決のためにはrelationshipにクラス解決をするための関数を渡してやれば良い。

base.py

from typing import Callable, Type

from sqlalchemy import FromClause
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

def from_clause(fn: Callable[[], Type[DeclarativeBase]]) -> Callable[[], FromClause]:
    return lambda: fn().__table__

def resolve_association_table():
    from .association import Association
    return Association

def resolve_child_table():
    from .child import Child
    return Child

parent.py

from typing import TYPE_CHECKING

from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

from .base import Base, resolve_association_table, from_clause, resolve_child_table

if TYPE_CHECKING:
    from .child import Child
    from .association import Association

class Parent(Base):
    __tablename__ = "left_table"

    id: Mapped[int] = mapped_column(primary_key=True)

    children: Mapped[list["Child"]] = relationship(
        resolve_child_table, # 参照先クラスを返す関数を渡す
        # secondaryはFromClauseを返す関数を渡す必要がある
        secondary=from_clause(resolve_association_table), back_populates="parents"
    )

    child_associations: Mapped[list["Association"]] = relationship(
        resolve_association_table,
        back_populates="parent"
    )

relationshipは関数を渡すことによってその関数の戻り値のテーブルクラスを参照先クラスとして認識してくれる。
関数として渡すことで、importが遅延されて循環importも起きないのがポイント。
またsecondaryとして指定する参照先のクラス情報も必要だが、こちらはテーブルクラスではなくFromClauseというオブジェクトを取る必要があるので一工夫必要。

これらの変更を各クラスに行うことでモデルを複数ファイルに分けつつもエラーに遭遇せず実行できる。
実際は共依存にあるクラスの何処か、例えばParentのみ、で上の変更を行えば各ファイルimportの結果が累積して残っているせいか知らないがエラーはでない。
ただ、一つのクラスだけで変更を行って解決するのは予期せぬエラーに合う可能性もあるのですべてのクラスで変更を実施するのが良いと思われる。

最後に全コード

base.py

from typing import Callable, Type

from sqlalchemy import FromClause
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

def from_clause(fn: Callable[[], Type[DeclarativeBase]]) -> Callable[[], FromClause]:
    return lambda: fn().__table__

def resolve_association_table():
    from .association import Association
    return Association

def resolve_parent_table():
    from .parent import Parent
    return Parent

def resolve_child_table():
    from .child import Child
    return Child

parent.py

from typing import TYPE_CHECKING

from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

from .base import Base, resolve_association_table, from_clause, resolve_child_table

if TYPE_CHECKING:
    from .child import Child
    from .association import Association

class Parent(Base):
    __tablename__ = "left_table"

    id: Mapped[int] = mapped_column(primary_key=True)

    children: Mapped[list["Child"]] = relationship(
        resolve_child_table,
        secondary=from_clause(resolve_association_table), back_populates="parents"
    )

    child_associations: Mapped[list["Association"]] = relationship(
        resolve_association_table,
        back_populates="parent"
    )

association.py

from typing import TYPE_CHECKING, Optional
from .base import Base, resolve_parent_table, resolve_child_table

from sqlalchemy.schema import ForeignKey
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

if TYPE_CHECKING:
    from .child import Child
    from .parent import Parent

class Association(Base):
    __tablename__ = "association_table"

    left_id: Mapped[int] = mapped_column(ForeignKey("left_table.id"), primary_key=True)
    right_id: Mapped[int] = mapped_column(
        ForeignKey("right_table.id"), primary_key=True
    )
    extra_data: Mapped[Optional[str]]

    child: Mapped["Child"] = relationship(resolve_child_table, back_populates="parent_associations")

    parent: Mapped["Parent"] = relationship(resolve_parent_table, back_populates="child_associations")

from typing import TYPE_CHECKING
from .base import Base, resolve_parent_table, from_clause, resolve_association_table

from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

if TYPE_CHECKING:
    from .association import Association
    from .parent import Parent

class Child(Base):
    __tablename__ = "right_table"

    id: Mapped[int] = mapped_column(primary_key=True)

    parents: Mapped[list["Parent"]] = relationship(
        resolve_parent_table,
        secondary=from_clause(resolve_association_table), back_populates="children"
    )

    parent_associations: Mapped[list["Association"]] = relationship(
        resolve_association_table,
        back_populates="child"
    )