SQLAlchemyでクラス継承とJoint Table Inheritanceを組み合わせた時にハマったこと

#@declared_attrは派生テーブルまで効果を持つ
##前提

• Joint Table Inheritanceで構成したい二つのテーブル(=モデル)がある(Employee, Manager, Engineer)。
• 大元の方(Employee)も他のテーブルと共通のカラムがあり、インデックスも含めてDRYに定義したい(Human)。
• 従って構成としてはHuman -> Employee -> (Manager, Engineer) となる。

##ハマったこと
Humanクラスに@declared_attrデコレータでインデックスを定義すると、一番下のクラスまでインデックスを張りに行く。

joint_ng.py

from sqlalchemy.ext.declarative import *
from sqlalchemy import *
from sqlalchemy_utils import database_exists, drop_database, create_database
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

class Human(object):
    age = Column(Integer)
    name = Column(String(50))

    @declared_attr
    def __table_args__(cls):
        return(Index('index_name', 'name'),)

class Employee(Human, Base):
    __tablename__ = 'employee'
    id = Column(Integer, primary_key=True)
    type = Column(String(50))

    __mapper_args__ = {
        'polymorphic_identity':'employee',
        'polymorphic_on':type
    }

class Engineer(Employee):
    __tablename__ = 'engineer'
    id = Column(Integer, ForeignKey('employee.id'), primary_key=True)
    engineer_name = Column(String(30))

    __mapper_args__ = {
        'polymorphic_identity':'engineer',
    }
    __table_args__ = {
        
    }

class Manager(Employee):
    __tablename__ = 'manager'
    id = Column(Integer, ForeignKey('employee.id'), primary_key=True)
    manager_name = Column(String(30))

    __mapper_args__ = {
        'polymorphic_identity':'manager',
    }
    __table_args__ = {
        
    }

engine = create_engine('mysql://user:pass@localhost/dummy?charset=utf8')

if database_exists(engine.url):
    drop_database(engine.url)
create_database(engine.url)
Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

m = Manager(age=10, name = 'foo', manager_name='hoge')
session.add(m)
session.commit()

nameカラムを定義していないManagerテーブルでnameカラムにインデックスを張ろうとしてエラーになる。

Traceback (most recent call last):
  File "joint.py", line 27, in <module>
    class Engineer(Employee):
(省略…)
  File "/home/satosi/.pyenv/versions/3.6.1/lib/python3.5/site-packages/sqlalchemy/util/_collections.py", line 194, in __getitem__
    return self._data[key]
KeyError: 'name'

当たり前と言えばそうなんだが、エラーメッセージからは状況が飲み込めず、原因を絞り込むのに手間取ったので公開する次第です。

##対策

単純にManagerクラスで__table_args__を再定義してやれば良い。

joint_ok.py

class Engineer(Employee):
    __tablename__ = 'engineer'
    id = Column(Integer, ForeignKey('employee.id'), primary_key=True)
    engineer_name = Column(String(30))

    __mapper_args__ = {
        'polymorphic_identity':'engineer',
    }
    __table_args__ = {
        
    }

class Manager(Employee):
    __tablename__ = 'manager'
    id = Column(Integer, ForeignKey('employee.id'), primary_key=True)
    manager_name = Column(String(30))

    __mapper_args__ = {
        'polymorphic_identity':'manager',
    }
    __table_args__ = {
        
    }

#polymorphic_identityの継承は想定外
##前提
テーブル定義とモデルのロジックを別々に書きたい。

##ハマったこと
declarative_base->Columnメンバーを指定するクラス(Manager)->アプリ層のクラス(SubManager)の順で継承させてみたが、ポリモーフィズムの要であるtypeが指定されない。

joint_ng2.py

import sys
from sqlalchemy.ext.declarative import *
from sqlalchemy import *
from sqlalchemy_utils import database_exists, drop_database, create_database
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

class Employee(Base):
    __tablename__ = 'employee'
    id = Column(Integer, primary_key=True)
    type = Column(String(50))

    __mapper_args__ = {
        'polymorphic_identity':'employee',
        'polymorphic_on':type
    }

class Manager(Employee):
    __tablename__ = 'manager'
    id = Column(Integer, ForeignKey('employee.id'), primary_key=True)
    manager_name = Column(String(30))
    
    __mapper_args__ = {
        'polymorphic_identity':'manager',
    }

    __table_args__ = {        
    }

    def shout(self):
        print('Oh')

class SubManager(Manager):
    def shout(self):
        print('Wah')

engine = create_engine('mysql://user:pass@localhost/dummy?charset=utf8')

if database_exists(engine.url):
    drop_database(engine.url)
create_database(engine.url)
Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

m = SubManager(manager_name='hoge')
m.shout()
session.add(m)
session.commit()
query = session.query(Manager)
print(query)
manager = query.first()
assert manager.type is None
manager.shout()

出力

Wah
SELECT manager.id AS manager_id, employee.id AS employee_id, employee.type AS employee_type, manager.manager_name AS manager_manager_name 
FROM employee INNER JOIN manager ON employee.id = manager.id
Oh

そこで@declared_attrを指定してみる。

joint_ng2.py

class Manager(Employee):
    (snip...)
    @declared_attr
    def __mapper_args__(cls):
       return {
        'polymorphic_identity':'manager',
       }

上手く行ったが二重定義だと警告される。

出力

/home/satosi/.pyenv/versions/3.6.1-mtxweb/lib/python3.5/site-packages/sqlalchemy/orm/mapper.py:1034: SAWarning: Reassigning polymorphic association for identity 'manager' from <Mapper at 0x7fb0202ed978; Manager> to <Mapper at 0x7fb0202edbe0; SubManager>: Check for duplicate use of 'manager' as value for polymorphic_identity.
  self, self.polymorphic_identity)
Wah
SELECT manager.id AS manager_id, employee.id AS employee_id, employee.type AS employee_type, manager.manager_name AS manager_manager_name 
FROM employee INNER JOIN manager ON employee.id = manager.id
Wah

##対策

単純に継承の順番を入れ替えれば良い。アプリ層とJoint Table Inheritanceの派生型を一対一で対応させている限り、機能性に違いはないはず。

joint_ok.py

from sqlalchemy.ext.declarative import *
from sqlalchemy import *
from sqlalchemy_utils import database_exists, drop_database, create_database
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

class _Manager():
    def shout(self):
        print('Oh')

class Employee(Base):
    __tablename__ = 'employee'
    id = Column(Integer, primary_key=True)
    type = Column(String(50))

    __mapper_args__ = {
        'polymorphic_identity':'employee',
        'polymorphic_on':type
    }

class Manager(Employee, _Manager):
    __tablename__ = 'manager'
    id = Column(Integer, ForeignKey('employee.id'), primary_key=True)
    manager_name = Column(String(30))
    
    __mapper_args__ ={
    'polymorphic_identity':'manager',
    }

    __table_args__ = {        
    }

    def shout(self):
        print('Wah')

engine = create_engine('mysql://user:pass@localhost/dummy?charset=utf8')

if database_exists(engine.url):
    drop_database(engine.url)
create_database(engine.url)
Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

m = Manager(manager_name='hoge')
m.shout()
session.add(m)
session.commit()
query = session.query(Manager)
print(query)
manager = query.first()
assert manager.type is not None
manager.shout()

出力

Wah
SELECT manager.id AS manager_id, employee.id AS employee_id, employee.type AS employee_type, manager.manager_name AS manager_manager_name 
FROM employee INNER JOIN manager ON employee.id = manager.id
Wah

お粗末様でした。どなたかのお役に立てば望外の喜びです。