今回は SQLite3 のインメモリデータベースをテストで使うときに生じる問題点と、その解決策について。
SQLite3 のインメモリデータベースを使うと、追加でソフトウェアをインストールしたり、データベースファイルを作ることなくリレーショナルデータベースを扱うことができる。
この点はリレーショナルデータベースを扱うソフトウェアを作るときに自動テストを組む上で望ましい特性といえる。
ただ、SQLite3 のインメモリデータベースには、制約が複数ある。
まず一つ目はコネクションを閉じると永続化した内容が消えてしまうところ。
そして、二つ目は異なるスレッドから内容を参照できないところ。
これらの解決策として、無理に制約のあるインメモリデータベースを使わずテンポラリファイルを使ってディスクに永続化することを提案する。
使った環境は次の通り。
$ sw_vers
ProductName: Mac OS X
ProductVersion: 10.14.3
BuildVersion: 18D109
$ python -V
Python 3.7.2
インメモリデータベースの問題点について
まずはインメモリデータベースを使うときの問題点について書いていく。
Python のインタプリタを起動しておこう。
$ python
コネクションを閉じるとデータが消えてしまう問題
最初に、SQLite3 のインメモリデータベースへのコネクションを作成する。
>>> import sqlite3
>>> connection = sqlite3.connect(':memory:')
コネクションからカーソルオブジェクトを生成する。
>>> cursor = connection.cursor()
カーソルオブジェクトを使うことで SQL 文を実行できる。
まずはテーブルの作成から。
>>> create_query = """
... CREATE TABLE users (
... id INTEGER,
... age INTEGER NOT NULL,
... name TEXT NOT NULL,
... PRIMARY KEY (id)
... );
... """
>>> cursor.execute(create_query)
<sqlite3.Cursor object at 0x1037fa260>
続いて、作成したテーブルにレコードを追加する。
>>> insert_query = """
... INSERT INTO users
... VALUES
... (1, 20, 'Alice'),
... (2, 30, 'Bob'),
... (3, 40, 'Carol');
... """
>>> cursor.execute(insert_query)
<sqlite3.Cursor object at 0x1037fa260>
この状態で、テーブルを SELECT する SQL 文を発行してみよう。
すると、ちゃんと永続化された内容が取得できる。
>>> select_query = """
... SELECT
... *
... FROM users
... """
>>> cursor.execute(select_query)
<sqlite3.Cursor object at 0x1037fa260>
>>> cursor.fetchall()
[(1, 20, 'Alice'), (2, 30, 'Bob'), (3, 40, 'Carol')]
しかし、ここで一旦コネクションを閉じるとどうなるだろうか？
>>> connection.commit()
>>> connection.close()
改めてインメモリデータベースへのコネクションを開いてみよう。
>>> connection = sqlite3.connect(':memory:')
>>> cursor = connection.cursor()
そして SELECT 文を発行する。
すると、テーブルがないというエラーになってしまった。
>>> cursor.execute(select_query)
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
sqlite3.OperationalError: no such table: users
このように、インメモリデータベースでは同じコネクションの中でしか永続化した内容が参照できない。
ちなみに、この問題は何もコネクションを閉じなければ問題ないかというと、そういう話でもない。
別のコネクションとして開いた場合にも問題になる。
まずはあるコネクションでテーブルを作ってレコードを追加しておく。
>>> conn1 = sqlite3.connect(':memory:')
>>> cur1 = conn1.cursor()
>>> create_query = """
... CREATE TABLE users (
... id INTEGER,
... age INTEGER NOT NULL,
... name TEXT NOT NULL,
... PRIMARY KEY (id)
... );
... """
>>> cur1.execute(create_query)
<sqlite3.Cursor object at 0x105c8b260>
>>> insert_query = """
... INSERT INTO users
... VALUES
... (1, 20, 'Alice'),
... (2, 30, 'Bob'),
... (3, 40, 'Carol');
... """
>>> cur1.execute(insert_query)
<sqlite3.Cursor object at 0x105c8b260>
そして、別のコネクションを開いて SELECT 文を発行してみる。
>>> conn2 = sqlite3.connect(':memory:')
>>> cur2 = conn2.cursor()
>>> select_query = """
... SELECT
... *
... FROM users
... """
>>> cur2.execute(select_query)
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
sqlite3.OperationalError: no such table: users
すると、やはりテーブルがないとエラーになってしまう。
異なるスレッド間でデータベースを参照できない問題
続いては異なるスレッド間でデータベースを参照できない問題について。
まずは、先ほどと同じように動作確認用のデータベースを作成しておく。
>>> connection = sqlite3.connect(':memory:')
>>> cursor = connection.cursor()
>>> create_query = """
... CREATE TABLE users (
... id INTEGER,
... age INTEGER NOT NULL,
... name TEXT NOT NULL,
... PRIMARY KEY (id)
... );
... """
>>> cursor.execute(create_query)
<sqlite3.Cursor object at 0x1037fa260>
>>> insert_query = """
... INSERT INTO users
... VALUES
... (1, 20, 'Alice'),
... (2, 30, 'Bob'),
... (3, 40, 'Carol');
... """
>>> cursor.execute(insert_query)
<sqlite3.Cursor object at 0x1037fa260>
続いて、次のようにデータベースのコネクションからレコードを SELECT する関数を定義する。
>>> def select(connection):
... cursor = connection.cursor()
... select_query = """
... SELECT
... *
... FROM users
... """
... cursor.execute(select_query)
... print(cursor.fetchall())
...
上記の関数を別のスレッドから実行する。
>>> import threading
>>> t = threading.Thread(target=select, args=(connection,))
すると、次のようにエラーになる。
SQLite のオブジェクトは作成したスレッドでしか扱ってはいけない、という旨のメッセージが表示されている。
しかし、異なるコネクションを開いても内容が参照できないのは先ほど見た通り。
>>> t.start()
>>> Exception in thread Thread-1:
Traceback (most recent call last):
File "/usr/local/Cellar/python/3.7.2_2/Frameworks/Python.framework/Versions/3.7/lib/python3.7/threading.py", line 917, in _bootstrap_inner
self.run()
File "/usr/local/Cellar/python/3.7.2_2/Frameworks/Python.framework/Versions/3.7/lib/python3.7/threading.py", line 865, in run
self._target(*self._args, **self._kwargs)
File "<stdin>", line 2, in select
sqlite3.ProgrammingError: SQLite objects created in a thread can only be used in that same thread. The object was created in thread id 4478109120 and this is thread id 123145389289472.
テンポラリファイルを使った解決策について
続いては、ここまで見てきた問題の解決策について提案する。
具体的には、無理に成約あるインメモリデータベースを使うのではなく、テンポラリファイルを使ってディスクに永続化するというもの。
Python には、使い終わった際に自動で削除されるファイルを扱うためのモジュールとして tempfile が用意されている。
まずは、このモジュールをインポートする。
>>> import tempfile
名前付きテンポラリファイルのオブジェクトをインスタンス化しよう。
>>> tfile = tempfile.NamedTemporaryFile()
このオブジェクトは、作成された時点でファイルシステムのどこかに対応するファイルが作成される。
>>> tfile.name
'/var/folders/1f/2k50hyvd2xq2p4yg668_ywsh0000gn/T/tmpc08x657m'
別のターミナルから確認すると、ちゃんとファイルができていることが分かる。
$ ls -alF /var/folders/1f/2k50hyvd2xq2p4yg668_ywsh0000gn/T/tmpc08x657m
-rw------- 1 amedama staff 0 3 3 18:36 /var/folders/1f/2k50hyvd2xq2p4yg668_ywsh0000gn/T/tmpc08x657m
このファイルは、オブジェクトが GC されると自動的に削除される。
>>> del tfile
次のように、ファイルがなくなった。
$ ls -alF /var/folders/1f/2k50hyvd2xq2p4yg668_ywsh0000gn/T/tmpc08x657m
ls: /var/folders/1f/2k50hyvd2xq2p4yg668_ywsh0000gn/T/tmpc08x657m: No such file or directory
続いては、このテンポラリファイルを使って SQLite3 の動作を確認してみよう。
テンポラリファイルのパスを使って SQLite3 のオブジェクトを作成する。
>>> tfile = tempfile.NamedTemporaryFile()
>>> connection = sqlite3.connect(tfile.name)
>>> cursor = connection.cursor()
データベースにテーブルを作って、レコードを追加する。
>>> cursor.execute(create_query)
<sqlite3.Cursor object at 0x105d1f0a0>
>>> cursor.execute(insert_query)
<sqlite3.Cursor object at 0x105d1f0a0>
終わったらコネクションを閉じてしまおう。
>>> connection.commit()
>>> connection.close()
その上で、テンポラリファイルのパスを使って再度コネクションを開く。
>>> connection = sqlite3.connect(tfile.name)
>>> cursor = connection.cursor()
そして SELECT 文を発行すると、ちゃんと先ほど永続化した内容が取れることが分かる。
>>> cursor.execute(select_query)
<sqlite3.Cursor object at 0x105edd6c0>
>>> cursor.fetchall()
[(1, 20, 'Alice'), (2, 30, 'Bob'), (3, 40, 'Carol')]
テンポラリファイルを使い終わって削除すれば、永続化に使っていた SQLite3 のデータベースファイルは自動的に削除される。
>>> del tfile
実際にテストを書くときのイメージ
もう少し具体的にテストを書くときのイメージをつけるために pytest を使った例を示す。
まずは pytest をインストールしておく。
$ pip install pytest
以下は SQLite3 を使ってリレーショナルデータベースの動作を確認するテストのサンプルコード。
import tempfile
import sqlite3
import pytest
@pytest.fixture(scope='module', autouse=True)
def temp_dbfile():
dbfile = tempfile.NamedTemporaryFile()
conn = sqlite3.connect(dbfile.name)
c = conn.cursor()
create_query = """
CREATE TABLE users (
id INTEGER,
age INTEGER NOT NULL,
name TEXT NOT NULL,
PRIMARY KEY (id)
);
"""
c.execute(create_query)
insert_query = """
INSERT INTO users VALUES
(1, 20, 'Alice'),
(2, 30, 'Bob'),
(3, 40, 'Carol');
"""
c.execute(insert_query)
conn.commit()
conn.close()
return dbfile
def test_example(temp_dbfile):
conn = sqlite3.connect(temp_dbfile.name)
c = conn.cursor()
select_query = """
SELECT
*
FROM users
WHERE
age > 35
"""
c.execute(select_query)
fetched_users = c.fetchall()
assert [(3, 40, 'Carol')] == fetched_users
if __name__ == '__main__':
pytest.main(['-v', __file__])
上記の実行結果は次の通り。
$ py.test -v test_example.py
============================= test session starts ==============================
platform darwin -- Python 3.7.2, pytest-4.3.0, py-1.8.0, pluggy-0.9.0 -- /Users/amedama/.virtualenvs/py37/bin/python3.7
cachedir: .pytest_cache
rootdir: /Users/amedama/Documents/temporary, inifile:
collected 1 item
test_example.py::test_example PASSED [100%]
=========================== 1 passed in 0.04 seconds ===========================
うまくテストがパスした。
注意事項
ここまで見てきた通り SQLite3 を使うとリレーショナルデータベースの自動テストが書きやすいのは、確かにそう。
ただ、リレーショナルデータベースは使うソフトウェアによってクセが強いので、結果をあまり信用しすぎない方が良い。
例えば、SQLite3 の自動テストで動作確認しながら開発していたけど、本番で使うデータベースと結合したときに動かないみたいなことが起こりうる。
これは、SQLAlchemy のような O/R マッパーを間に挟んでいても同じ。
そのため、本番で MySQL や PostgreSQL を使うのであれば、SQLite3 の自動テストはあくまで軽い動作確認程度にとどめた方が良い。
早めに Docker などを用いて本番で使うデータベースを使った E2E テストを書いておくと、だいぶ心理的には安心できる。
いじょう。
