SQLAlchemy ORMの宣言的モデル定義でMany to Manyのrelationshipがある際にモデルファイルを複数に分ける
この記事で書くこと
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ファイルで定義するのは大変なのでこれらは分割したくなる。
複数ファイルにモデル定義を分ける際の問題と解決方法
最初のファイル分け
ナイーブにファイル分割すると以下のようになる。
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
pass
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"
)
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"
)
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を防ぐ事ができる。
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にクラス解決をするための関数を渡してやれば良い。
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
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の結果が累積して残っているせいか知らないがエラーはでない。
ただ、一つのクラスだけで変更を行って解決するのは予期せぬエラーに合う可能性もあるのですべてのクラスで変更を実施するのが良いと思われる。
最後に全コード
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
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"
)
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"
)