Python: SQLAlchemy のテーブルに後からインデックスを追加する

今回は、Python の O/R マッパーである SQLAlchemy について。テーブルを定義した時点のモデルには無かったインデックスを、後から追加する方法についてメモしておく。

なお、実務における RDBMS のスキーマ変更に関しては、Alembic のようなフレームワークを使ってバージョン管理することを強くおすすめしたい。

blog.amedama.jp

使った環境は次のとおり。

$ sw_vers                     
ProductName:    Mac OS X
ProductVersion: 10.15.7
BuildVersion:   19H2
$ python -V
Python 3.8.6
$ python -c "import sqlite3; print(sqlite3.version)"    
2.6.0
$ sqlite3 --version           
3.28.0 2019-04-15 14:49:49 378230ae7f4b721c8b8d83c8ceb891449685cd23b1702a57841f1be40b5daapl

下準備

あらかじめ、SQLAlchemy をインストールしておく。

$ pip install sqlalchemy

最初からテーブルの定義にインデックスがある場合

はじめに、最初からテーブルを表すモデルにインデックスの指定がある場合について確認しておく。以下のサンプルコードには User というクラスが users というテーブルを表している。そして、age というカラムに対応するアトリビュートに index=True が指定されている。こうなっていると、テーブルを初期化した時点で、age というカラムにインデックスが有効となる。

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

import logging

from sqlalchemy import create_engine
from sqlalchemy.engine import Engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.schema import Column
from sqlalchemy.types import Integer
from sqlalchemy.types import Text

ModelBase = declarative_base()


LOGGER = logging.getLogger(__name__)


class User(ModelBase):
    """ユーザのデータを模したサンプル用のモデル"""
    __tablename__ = 'users'

    # ユーザ名
    name = Column(Text, primary_key=True)
    # NOTE: 最初からインデックスを張るときは index=True オプションを使う
    age = Column(Integer, nullable=False, index=True)


def init_db(engine: Engine, drop: bool = False):
    """データベースを初期化する"""
    if drop:
        ModelBase.metadata.drop_all(engine)

    ModelBase.metadata.create_all(engine)


def get_engine(db_uri: str) -> Engine:
    engine = create_engine(
        db_uri,
        pool_recycle=3600,
        encoding='utf-8',
    )
    return engine


def main():
    # SQLAlchemy のログを DEBUG レベルで出力する
    logger = logging.getLogger('sqlalchemy')
    logger.setLevel(logging.DEBUG)
    # DEBUG レベル以上のログを出力する
    logging.basicConfig(level=logging.DEBUG)

    # データベースを初期化する
    db_uri = 'sqlite:///example.db?cache=shared'
    engine = get_engine(db_uri)
    init_db(engine, drop=True)


if __name__ == '__main__':
    main()

上記を実行してみよう。 SQLAlchemy のデバッグログを有効にしているため、バックエンドに発行したクエリなどが出力される。ここでは、それらの出力は省略している。

$ python withindex.py
...

作成されたデータベースのスキーマを確認してみよう。すると、次のとおり age カラムにインデックスが確認できる。

$ sqlite3 example.db ".schema"            
CREATE TABLE sqlite_sequence(name,seq);
CREATE TABLE users (
    name TEXT NOT NULL, 
    age INTEGER NOT NULL, 
    PRIMARY KEY (name)
);
CREATE INDEX ix_users_age ON users (age);

最初からテーブルの定義にインデックスがない場合

続いてが本題の、最初の定義にはインデックスの指定がない場合について。以下のサンプルコードでは、先ほど作成したテーブルを DROP して、新たにインデックスの指定がないものを作り直している。

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

import logging

from sqlalchemy import create_engine
from sqlalchemy.engine import Engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.schema import Column
from sqlalchemy.types import Integer
from sqlalchemy.types import Text

ModelBase = declarative_base()


LOGGER = logging.getLogger(__name__)


class User(ModelBase):
    __tablename__ = 'users'

    name = Column(Text, primary_key=True)
    # この時点ではインデックスがない
    age = Column(Integer, nullable=False)


def init_db(engine: Engine, drop: bool = False):
    if drop:
        ModelBase.metadata.drop_all(engine)

    ModelBase.metadata.create_all(engine)


def get_engine(db_uri: str) -> Engine:
    engine = create_engine(
        db_uri,
        pool_recycle=3600,
        encoding='utf-8',
    )
    return engine


def main():
    logger = logging.getLogger('sqlalchemy')
    logger.setLevel(logging.DEBUG)
    logging.basicConfig(level=logging.DEBUG)

    db_uri = 'sqlite:///example.db?cache=shared'
    engine = get_engine(db_uri)
    init_db(engine, drop=True)


if __name__ == '__main__':
    main()

上記を実行してみよう。

$ python withoutindex.py
...

すると、次のようにインデックスのないテーブルが作り直される。

$ sqlite3 example.db ".schema"
CREATE TABLE users (
    name TEXT NOT NULL, 
    age INTEGER NOT NULL, 
    PRIMARY KEY (name)
);

この状況から、ただモデルのアトリビュートに index=True を追加して ModelBase.metadata.create_all() しても上手くいかない。

SQLAlchemy でインデックスを後から追加する

SQLAlchemy で後からインデックスを追加するには Index というクラスのインスタンスを作る。そして、create() というメソッドを呼ぶことでインデックスが作成できるようだ。

文章で説明するよりも実際に試す方が分かりやすいと思うので REPL を使って作業していく。なお、REPL を起動する場所は、先ほどのサンプルコードを作ったディレクトリにしよう。あるいは、サンプルコードの場所をシェル変数の PYTHONPATH に指定してもいい。

$ python

以下がインデックスを作ることができるクラス。

>>> from sqlalchemy import Index

既存モデルのカラムに対応したインデックスを作りたいなら、次のようにする。先ほどサンプルコードとして示したモジュールをインポートして使っている。

>>> from withoutindex import User
>>> UserAgeIndex = Index('ix_users_age', User.age)

動かすのに sqlalchemy.engine.Engine のインスタンスが必要なので用意する。

>>> from withoutindex import get_engine
>>> db_uri = 'sqlite:///example.db?cache=shared'
>>> engine = get_engine(db_uri)

あとは、先ほど作った Index クラスのインスタンスに対して create() メソッドの呼ぶだけ。

>>> UserAgeIndex.create(engine)
Index('ix_users_age', Column('age', Integer(), table=<users>, nullable=False))

テーブルの定義を確認すると、次のようにインデックスが追加されている。

$ sqlite3 example.db ".schema"
CREATE TABLE users (
    name TEXT NOT NULL, 
    age INTEGER NOT NULL, 
    PRIMARY KEY (name)
);
CREATE INDEX ix_users_age ON users (age);

いじょう。

繰り返しになるけど、実際には上記のような手動オペレーションをするのではなく、Alembic などを使ってスキーマをバージョン管理しよう。