Python: テストで SQLite3 のインメモリデータベースを使うときの問題点と解決策

今回は 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 を使ってリレーショナルデータベースの動作を確認するテストのサンプルコード。

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

import tempfile
import sqlite3

import pytest


@pytest.fixture(scope='module', autouse=True)
def temp_dbfile():
    # テンポラリファイルを用意する
    dbfile = tempfile.NamedTemporaryFile()

    # SQLite3 のデータベースファイルとして使う
    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 テストを書いておくと、だいぶ心理的には安心できる。

いじょう。