Python: SQLAlchemy の生成する SQL をテストするパッケージを作ってみた

SQLAlchemy は Python でよく使われている O/R マッパーの一つ。今回は、そんな SQLAlchemy が生成する SQL 文を確認するためのパッケージを作ってみたよ、という話。

具体的には、以下の sqlalchemy-profile というパッケージを作ってみた。このエントリでは、なんでこんなものを作ったのかみたいな話をしてみる。

使った環境は次の通り。ただし sqlalchemy-profile 自体はプラットフォームに依存せず Python 2.7, 3.3 ~ 3.6 に対応している。

$ sw_vers 
ProductName:    Mac OS X
ProductVersion: 10.12.4
BuildVersion:   16E195
$ python --version
Python 3.6.1

O/R マッパーについて

O/R マッパーというのは、プログラミング言語からリレーショナルデータベース (RDB) を良い感じに使うための機能ないしライブラリの総称。プログラミング言語から RDB を操作するための SQL 文を直に扱ってしまうと、両者のパラダイムの違いから色々な問題が起こる。この問題は、一般にインピーダンスミスマッチと呼ばれている。そこで登場するのが O/R マッパーで、これを使うとプログラミング言語のオブジェクトを操作する形で RDB を操作できるようになる。

論よりソースということで、まずは SQLAlchemy の基本的な使い方から見てみよう。その前に SQLAlchemy 自体をインストールしておく。

$ pip install sqlalchemy

そして次に示すのがサンプルコード。ユーザ情報を模したモデルクラスを用意して、それを SQLite のオンメモリデータベースで永続化している。この中には SQL 文が全く登場していないところがポイントとなる。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from sqlalchemy.ext import declarative
from sqlalchemy import Column
from sqlalchemy import Integer
from sqlalchemy import Text
from sqlalchemy import create_engine
from sqlalchemy.orm.session import sessionmaker

Base = declarative.declarative_base()


class User(Base):
    """SQLAlchemy のモデルクラス

    このクラスが RDB のテーブルと対応し、インスタンスはテーブルの一レコードに対応する
    ここではユーザの情報を格納するテーブルを模している"""
    __tablename__ = 'users'

    # テーブルの主キー
    id = Column(Integer, primary_key=True)
    # 名前を入れるカラム
    name = Column(Text, nullable=False)


def main():
    # データベースとの接続に使う情報
    # ここでは SQLite のオンメモリデータベースを使う
    # echo=True とすることで生成される SQL 文を確認できる
    engine = create_engine('sqlite:///', echo=True)
    # モデルの情報を元にテーブルを生成する
    Base.metadata.create_all(engine)
    # データベースとのセッションを確立する
    session_maker = sessionmaker(bind=engine)
    session = session_maker()

    # データベースのトランザクションを作る
    with session.begin(subtransactions=True):
        # レコードに対応するモデルのインスタンスを作る
        user = User(name='Alice')
        # そのインスタンスをセッションに追加する
        session.add(user)

    # トランザクションがコミットされてオブジェクトが RDB で永続化される

if __name__ == '__main__':
    main()

上記のサンプルコードでは生成される SQL 文を標準出力に表示するようにしている。なので、実行するとこんな感じの出力が得られる。

2017-04-20 04:48:30,976 INFO sqlalchemy.engine.base.Engine SELECT CAST('test plain returns' AS VARCHAR(60)) AS anon_1
2017-04-20 04:48:30,976 INFO sqlalchemy.engine.base.Engine ()
2017-04-20 04:48:30,978 INFO sqlalchemy.engine.base.Engine SELECT CAST('test unicode returns' AS VARCHAR(60)) AS anon_1
2017-04-20 04:48:30,978 INFO sqlalchemy.engine.base.Engine ()
2017-04-20 04:48:30,980 INFO sqlalchemy.engine.base.Engine PRAGMA table_info("users")
2017-04-20 04:48:30,980 INFO sqlalchemy.engine.base.Engine ()
2017-04-20 04:48:30,982 INFO sqlalchemy.engine.base.Engine 
CREATE TABLE users (
    id INTEGER NOT NULL, 
    name TEXT NOT NULL, 
    PRIMARY KEY (id)
)


2017-04-20 04:48:30,983 INFO sqlalchemy.engine.base.Engine ()
2017-04-20 04:48:30,984 INFO sqlalchemy.engine.base.Engine COMMIT
2017-04-20 04:48:30,987 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2017-04-20 04:48:30,989 INFO sqlalchemy.engine.base.Engine INSERT INTO users (name) VALUES (?)
2017-04-20 04:48:30,989 INFO sqlalchemy.engine.base.Engine ('Alice',)

たしかに Python のオブジェクトを使うだけで RDB を操作できた。便利。ただし、上記で使った生成した SQL 文を出力する機能はデバッグ用途なので普段は無効にされる場合が多い。

SQL が隠蔽されることのメリットとデメリット

先ほど見た通り O/R マッパーを使うと Python のオブジェクトを通して RDB を操作できるようになる。これにはインピーダンスミスマッチの解消という多大なメリットがある反面、生成される SQL が隠蔽されるというデメリットもある。

例えば、直接 SQL を書くならそんな非効率なクエリは組まないよね・・・というような内容も、気をつけていないと生成されうる。これは、典型的には N + 1 問題とか。それを防ぐには、これまでだとコードから生成される SQL 文を推測したり、あるいは先ほどのようにして実際に目で見て確かめていた。慣れてくるとどんな SQL 文が発行されるか分かってくるのと、実際に目で見て確かめるのは手間なので大体は前者になっている。

ただ、パフォーマンスチューニングの世界では、推測する前に測定せよという格言もある。実際に生成される SQL 文を、ユニットテストで確認できるようになっているべきなのでは、という考えに至った。それが、今回作ったパッケージ sqlalchemy-profile のモチベーションになっている。

ただし、どんな SQL 文が生成されるかは SQLAlchemy のアルゴリズム次第なので、気をつけないとテストのメンテナンス性が低下する恐れはあると思う。これは、SQLAlchemy のバージョン変更とか、些細なモデルの構造変更でテストを修正する手間がかかるかも、ということ。とはいえ、それはそれで生成される SQL が変更されたことにちゃんと気づけるのは大事じゃないかという感じでいる。

sqlalchemy-profile について

やっと本題に入るんだけど、前述した問題を解消すべく sqlalchemy-profile という Python のパッケージを作ってみた。これを使うことで、SQLAlchemy が生成する SQL 文を確かめることができる。

Python のパッケージリポジトリである PyPI にも登録しておいた。 pypi.python.org

インストールは Python のパッケージマネージャの PIP からできる。

$ pip install sqlalchemy-profile

使い方

ここからは sqlalchemy-profile の具体的な使い方について見ていく。シンプルなのでサンプルコードをいくつか見れば、すぐに分かってもらえると思う。ちなみに、トラッキングしている SQL 文は今のところ INSERT, UPDATE, SELECT, DELETE の四つ。

以下のサンプルコードでは、最も基本的な使い方を示している。まず、プロファイラとなる StatementProfiler には SQLAlchemy のデータベースとの接続情報を渡す。そして、プロファイルしている期間中に実行された SQL 文を記録する、というもの。ユニットテストで利用することを意図しているので、サンプルコードも Python の unittest モジュールを使うものにしてみた。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import unittest

from sqlalchemy import create_engine

from sqlalchemy_profile import StatementProfiler


class Test_RawExecute(unittest.TestCase):

    def test(self):
        # データベースとの接続を確立する
        engine = create_engine('sqlite:///')
        connection = engine.connect()

        # データベースとの接続情報を渡してプロファイラをインスタンス化する
        profiler = StatementProfiler(engine)
        # プロファイルを開始する
        profiler.start()

        # SQLAlchemy を使って RDB を操作する
        # ここでは、サンプルコードをシンプルにする目的で低レイヤーな API を使っている
        connection.execute('SELECT 1')
        connection.execute('SELECT 2')

        # プロファイルを停止する
        profiler.stop()

        # 実行された SQL 文の内容を確認する
        assert profiler.count == 2
        assert profiler.select == 2


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

上記では、分かりやすくするためにあえて SQLAlchemy の直接 SQL 文を扱う低レイヤーな API を使っている。

上記を実行するとテストがパスする。

$ python profile101.py 
.
----------------------------------------------------------------------
Ran 1 test in 0.020s

OK

このとき assert しているところの数値を変更すると、当然だけどテストは失敗するようになる。想定していた SQL 文の数と、実際に発行された数が一致しないため。

$ python profile101.py
F
======================================================================
FAIL: test (__main__.Test_RawExecute)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "profile101.py", line 32, in test
    assert profiler.count == 1
AssertionError

----------------------------------------------------------------------
Ran 1 test in 0.017s

FAILED (failures=1)

O/R マッピングと共に使う

先ほどの例では、分かりやすさのためにあえて SQLAlchemy の直接 SQL 文を扱う低レイヤーな API を使っていた。もちろん sqlalchemy-profile は O/R マッピングをしたコードでも動作するし、使い方については何も変わらない。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import unittest

from sqlalchemy.ext import declarative
from sqlalchemy import Column
from sqlalchemy import Integer
from sqlalchemy import Text
from sqlalchemy import create_engine
from sqlalchemy.orm.session import sessionmaker

from sqlalchemy_profile import StatementProfiler

Base = declarative.declarative_base()


class _User(Base):
    """ユーザ情報を模したモデルクラス"""
    __tablename__ = 'users'

    # 主キー
    id = Column(Integer, primary_key=True)
    # 名前を格納するカラム
    name = Column(Text, nullable=False)


class Test_ORMapping(unittest.TestCase):

    def setUp(self):
        """テストが実行される前の下準備"""
        self.engine = create_engine('sqlite:///')
        Base.metadata.create_all(self.engine)
        self.session_maker = sessionmaker(bind=self.engine)

    def tearDown(self):
        """テストが実行された後の後始末"""
        Base.metadata.drop_all(self.engine)

    def test(self):
        session = self.session_maker()

        profiler = StatementProfiler(self.engine)
        profiler.start()

        # 以下のユーザを模したインスタンスを一通り CRUD していく
        user = _User(name='Alice')

        # INSERT
        with session.begin(subtransactions=True):
            session.add(user)

        # UPDATE
        with session.begin(subtransactions=True):
            user.name = 'Bob'

        # SELECT
        session.query(_User).all()

        # DELETE
        with session.begin(subtransactions=True):
            session.delete(user)

        profiler.stop()

        # SQL 文は各一回ずつ実行されているはず
        assert profiler.count == 4
        assert profiler.insert == 1
        assert profiler.update == 1
        assert profiler.select == 1
        assert profiler.delete == 1


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

with ステートメントと共に使う

これまでの例ではプロファイリング期間を start() メソッドと stop() メソッドで制御したけど、これは with でも代用できる。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import unittest

from sqlalchemy import create_engine

from sqlalchemy_profile import StatementProfiler


class Test_WithStatement(unittest.TestCase):

    def test(self):
        engine = create_engine('sqlite:///')
        connection = engine.connect()

        # with ステートメントのスコープで実行された SQL 文を記録する
        with StatementProfiler(engine) as profiler:
            connection.execute('SELECT 1')
            connection.execute('SELECT 2')

        assert profiler.count == 2
        assert profiler.select == 2


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

デコレータと共に使う

with を使うのもめんどくさいなー、というときはテストメソッド自体をデコレータで修飾しちゃうような使い方もできる。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import unittest

from sqlalchemy import create_engine

from sqlalchemy_profile import sqlprofile

ENGINE = create_engine('sqlite:///')


class Test_Decorator(unittest.TestCase):

    # ユニットテストのメソッドをデコレータで修飾する
    # メソッド内で実行されることが想定される SQL 文の数を指定する
    @sqlprofile(ENGINE, count=2, select=2)
    def test(self):
        connection = ENGINE.connect()

        connection.execute('SELECT 1')
        connection.execute('SELECT 2')


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

SQL 文の種類と順序まで確認したい

いやいや回数だけのアサーションとかアバウトすぎるでしょ、っていうときは StatementProfiler#sequence を使う。これで INSERT, UPDATE, SELECT, DELETE が、どんな順番で実行されたかを確認できる。中身は文字列で、それぞれの操作の頭文字が入っている。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import unittest

from sqlalchemy.ext import declarative
from sqlalchemy import Column
from sqlalchemy import Integer
from sqlalchemy import Text
from sqlalchemy import create_engine
from sqlalchemy.orm.session import sessionmaker

from sqlalchemy_profile import StatementProfiler

Base = declarative.declarative_base()


class _User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(Text, nullable=False)


class Test_ORMapping(unittest.TestCase):

    def setUp(self):
        self.engine = create_engine('sqlite:///')
        Base.metadata.create_all(self.engine)
        self.session_maker = sessionmaker(bind=self.engine)

    def tearDown(self):
        Base.metadata.drop_all(self.engine)

    def test(self):
        session = self.session_maker()

        profiler = StatementProfiler(self.engine)
        profiler.start()

        user = _User(name='Alice')

        # INSERT
        with session.begin(subtransactions=True):
            session.add(user)

        # UPDATE
        with session.begin(subtransactions=True):
            user.name = 'Bob'

        # SELECT
        session.query(_User).all()

        # DELETE
        with session.begin(subtransactions=True):
            session.delete(user)

        profiler.stop()

        # [I]NSERT -> [U]PDATE -> [S]ELECT -> [D]ELETE
        assert profiler.sequence == 'IUSD'


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

もちろんデコレータの API でも使える。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import unittest

from sqlalchemy import create_engine

from sqlalchemy_profile import sqlprofile

ENGINE = create_engine('sqlite:///')


class Test_Decorator(unittest.TestCase):

    # SELECT -> SELECT = SS
    @sqlprofile(ENGINE, seq='SS')
    def test(self):
        connection = ENGINE.connect()

        connection.execute('SELECT 1')
        connection.execute('SELECT 2')


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

もっと厳密にアサーションしたい

いやいや SQL 文の構造までもっと調べたいよ、というときは StatementProfiler#statements や StatementProfiler#statements_with_parameters を使う。

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import unittest

from sqlalchemy.ext import declarative
from sqlalchemy import Column
from sqlalchemy import Integer
from sqlalchemy import Text
from sqlalchemy import create_engine
from sqlalchemy.orm.session import sessionmaker

from sqlalchemy_profile import StatementProfiler

Base = declarative.declarative_base()


class _User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(Text, nullable=False)


class Test_ORMapping(unittest.TestCase):

    def setUp(self):
        self.engine = create_engine('sqlite:///')
        Base.metadata.create_all(self.engine)
        self.session_maker = sessionmaker(bind=self.engine)

    def tearDown(self):
        Base.metadata.drop_all(self.engine)

    def test(self):
        session = self.session_maker()

        profiler = StatementProfiler(self.engine)
        profiler.start()

        user = _User(name='Alice')

        # INSERT
        with session.begin(subtransactions=True):
            session.add(user)

        profiler.stop()

        assert profiler.count == 1
        assert profiler.insert == 1

        # 生の SQL 文を取得する
        print(profiler.statements)
        print(profiler.statements_with_parameters)


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

こんな感じ。

['INSERT INTO users (name) VALUES (?)']
[('INSERT INTO users (name) VALUES (?)', ('Alice',))]
.
----------------------------------------------------------------------
Ran 1 test in 0.019s

OK

こちらは、今のところデコレータの API では使えない。