Python SQLAlchemyでちょっとハマったこと

By On 2014/03/24 · Leave a Comment · In MySQL, Python

沼津で深海魚食えるって聞いて先週食ってきた。食うときに店の人に魚の名前聞いたけど、ゲホウとかアブラボウズとデン(だったかな？)以外は忘れた。いろいろ食ったけど全部覚えられん。深海魚は普通に食えることがわかった。美味いよ。でもまあ目の前にマグロとサーモンと深海魚があったら、たぶんマグロとサーモンを食うと思うw 記念として食うには良いかなw デンとゲホウは俺は好き。
ちょっとSQLAlchemyの練習がてら、遊びサイトやら社内ツールを作ってて、適当にやってたらハマった。平たく言うとsession.commit()時に発行されるSQLの順番は不定になるので、コード上はテーブルAにinsert→テーブルBにinsert→commitと書いていても、先にテーブルBにinsertが発行される場合がある。なので、外部キーを貼るときはちゃんとリレーションを使う。また、外部キーを貼っていなくてもテーブルの更新に順序性が必要な場合はsession.flush()を適切にしてあげないとダメ。マニュアル読んだときは「flush()はcommit()時にも実行される」みたいな記述があったから、commitで一括でいいやって思ってたが…。



モデル

例として、サッカーのチームと選手の関係イメージしてこんなの↓をモデルとして定義する。ここで、「#club = relation(‘Club’, backref=’player’, uselist=False)」を最初はコメントアウトしておく。冒頭にも書いた通り、relationをしてなくてハマったので。
(※ ところで話それるんだけど、clubテーブルにクラブIDみたいなidカラムがない点にツッこむ人ってやっぱ多いのかな？とりあえず先頭に「なんちゃらid」があるような設計を目にすることがあるんだけど、使わないならいらんだろ、と思っている。何でもかんでも先頭に「id」を振るのは好きじゃないんだよね。それにこの場合はチーム名で一意性を確保できるのであれば、それがそのままPKでいいんじゃね？って思う。playerには「Ronaldo」みたいな同名もあるので、この場合はレコードの一意性を確保するために「id」を設けるのが良いと思うが。それよりもvarcharがPKでいいのか？ってつっこみもされそうだけど。まあこれは遊びなので。)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
from sqlalchemy import create_engine, Column, ForeignKey
from sqlalchemy.dialects.mysql import (
    INTEGER,
    VARCHAR,
    TINYINT,
)
from sqlalchemy.orm import sessionmaker, relation
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Player(Base):
    __tablename__ = 'player'
    __table_args__ = {
        'mysql_engine':'InnoDB',
        'mysql_charset':'utf8',
    }

    id = Column('id', INTEGER(unsigned=True),
        primary_key=True,
        autoincrement=True)
    player_name = Column('player_name', VARCHAR(length=64),
        nullable=False)
    nationality = Column('nationality', VARCHAR(length=32),
        nullable=False)
    club = relation('Club', backref='player', uselist=False)

class Club(Base):
    __tablename__ = 'club'
    __table_args__ = {
        'mysql_engine':'InnoDB',
        'mysql_charset':'utf8',
    }

    club_name = Column('club_name', VARCHAR(length=64),
        primary_key=True,
        autoincrement=False)
    number = Column('number', TINYINT(unsigned=True),
        primary_key=True,
        autoincrement=False)
    country = Column('country', VARCHAR(length=32),
        nullable=False)
    player_id = Column('player_id', INTEGER(unsigned=True),
        ForeignKey('player.id'),
        unique=True,
        nullable=False)

player

1
2
3
4
5
6
CREATE TABLE `player` (
  `id` int(10) unsigned NOT NULL AUTO_INCREMENT,
  `player_name` varchar(64) NOT NULL,
  `nationality` varchar(32) NOT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8

データサンプル
id player_name nationality
1 KAZU Japan
2 Ronaldo Brazil
3 Ronaldo Portugal
4 Diego Armando Maradona Argentine
club

1
2
3
4
5
6
7
8
9
CREATE TABLE `club` (
  `club_name` varchar(64) NOT NULL,
  `number` tinyint(3) unsigned NOT NULL,
  `country` varchar(32) NOT NULL,
  `player_id` int(10) unsigned NOT NULL,
  PRIMARY KEY (`club_name`,`number`),
  UNIQUE KEY `player_id` (`player_id`),
  CONSTRAINT `club_ibfk_1` FOREIGN KEY (`player_id`) REFERENCES `player` (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8

データサンプル
club_name number country player_id
Verdy Kawasaki 11 Japan 1
Real Madrid 9 Spain 2
Real Madrid 7 Spain 3
Napoli 10 Italy 4
テストコード

で、テストコード。このテストがパスしたりコケたりする。コケるときの原因は外部キー制約違反。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
import unittest
from models import Base, Player, Club

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

dsn = 'mysql+pymysql://%s:%s@%s/%s?charset=%s' % ('root', '', 'localhost', 'test', 'utf8')

player_list = [
        ('KAZU', 'Japan'),
        ('Ronaldo', 'Brazil'),
        ('Ronaldo', 'Portugal'),
        ('Diego Armando Maradona', 'Argentine')
]
club_list = [
        ('Verdy Kawasaki', 11, 'Japan', 1),
        ('Real Madrid', 9, 'Spain', 2),
        ('Real Madrid', 7, 'Spain', 3),
        ('Napoli', 10, 'Italy', 4),
]

class TestSQLAlchemy(unittest.TestCase):

    def setUp(self):
        self.engine = create_engine(
            dsn,
            echo=True
        )
        Base.metadata.drop_all(self.engine)
        Base.metadata.create_all(self.engine)
        Session = sessionmaker(bind=self.engine)
        self.session = Session()

    def tearDown(self):
        self.session.commit()
        self.session.close()
        Base.metadata.drop_all(self.engine)

    def test_insert(self):
        for p in player_list:
            player = Player(player_name=p[0], nationality=p[1])
            self.session.add(player)
        for c in club_list:
            club = Club(club_name=c[0], number=c[1], country=c[2], player_id=c[3])
            self.session.add(club)

if __name__ == '__main__':
    unittest.main()

で、コケるときにクエリログを調査してみると最初にclubへのinsertが発行されていた。こちらの期待としてはplayerへのinsertが先に走って欲しいんだけど。どうやらplayerに先にinsertがされる場合もあれば、clubに先にinsertが発行される場合もあるようだ。

対処法その1 – 都度flush(もしくはコミット)

playerをaddしてflush→clubをaddしてflushとする。これで100%通る。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
import unittest
from models import Base, Player, Club

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

dsn = 'mysql+pymysql://%s:%s@%s/%s?charset=%s' % ('root', '', 'localhost', 'test', 'utf8')

player_list = [
        ('KAZU', 'Japan'),
        ('Ronaldo', 'Brazil'),
        ('Ronaldo', 'Portugal'),
        ('Diego Armando Maradona', 'Argentine')
]
club_list = [
        ('Verdy Kawasaki', 11, 'Japan', 1),
        ('Real Madrid', 9, 'Spain', 2),
        ('Real Madrid', 7, 'Spain', 3),
        ('Napoli', 10, 'Italy', 4),
]

class TestSQLAlchemy(unittest.TestCase):

    def setUp(self):
        self.engine = create_engine(
            dsn,
            echo=True
        )
        Base.metadata.drop_all(self.engine)
        Base.metadata.create_all(self.engine)
        Session = sessionmaker(bind=self.engine)
        self.session = Session()

    def tearDown(self):
        self.session.commit()
        self.session.close()
        Base.metadata.drop_all(self.engine)

    def test_insert_2(self):
        for p in player_list:
            player = Player(player_name=p[0], nationality=p[1])
            self.session.add(player)
        self.session.flush()
        for c in club_list:
            club = Club(club_name=c[0], number=c[1], country=c[2], player_id=c[3])
            self.session.add(club)
        self.session.flush()

if __name__ == '__main__':
    unittest.main()

対処法その2 – ちゃんとリレーションを貼る

この例の場合は外部キーを使っているので、ちゃんとリレーションを貼る。モデルの「#club = relation(‘Club’, backref=’player’, uselist=False)」のコメントアウトを外して、最初のテストコードを流す。100%通る。
また、せっかくリレーションを貼ったので、player.clubを使って次のテストコードとするのがいいだろう。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
import unittest
from models import Base, Player, Club

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

dsn = 'mysql+pymysql://%s:%s@%s/%s?charset=%s' % ('root', '', 'localhost', 'test', 'utf8')

player_list = [
        ('KAZU', 'Japan'),
        ('Ronaldo', 'Brazil'),
        ('Ronaldo', 'Portugal'),
        ('Diego Armando Maradona', 'Argentine')
]
club_list = [
        ('Verdy Kawasaki', 11, 'Japan'),
        ('Real Madrid', 9, 'Spain'),
        ('Real Madrid', 7, 'Spain'),
        ('Napoli', 10, 'Italy'),
]

class TestSQLAlchemy(unittest.TestCase):

    def setUp(self):
        self.engine = create_engine(
            dsn,
            echo=True
        )
        Base.metadata.drop_all(self.engine)
        Base.metadata.create_all(self.engine)
        Session = sessionmaker(bind=self.engine)
        self.session = Session()

    def tearDown(self):
        self.session.commit()
        self.session.close()
        Base.metadata.drop_all(self.engine)

    def test_insert_with_relation(self):
        for i, p in enumerate(player_list):
            player = Player(player_name=p[0], nationality=p[1])
            player.club = Club(club_name=club_list[i][0], number=club_list[i][1], country=club_list[i][2])
            self.session.add(player)

if __name__ == '__main__':
    unittest.main()

まとめ
  • 適切なタイミングでflushを発行する
  • 外部キー使うときはちゃんとリレーション使う
テストに使ったコードは↓に。


  • https://gist.github.com/hiroakis/9739453


おわり

 

コメントを残す

メールアドレスが公開されることはありません。 * が付いている欄は必須項目です

次のHTML タグと属性が使えます: <a href="" title=""> <abbr title=""> <acronym title=""> <b> <blockquote cite=""> <cite> <code> <del datetime=""> <em> <i> <q cite=""> <strike> <strong> <img localsrc="" alt="">