概要

mock-alchemyのUnifiedAlchemyMagicMockを使用してSQLAlchemyステートメントを実行してみる。

UnifiedAlchemyMagicMock とは

SQLALchemyを組み合わせてsessionをmockできる
一般的なSQLALchemy session関数を統合してassertionを簡単にできる

参考

環境構築

# Pythonのバージョンを確認
python --version
> Python 3.7.3

# mock-alchemyのインストール
> pip install mock-alchemy

ファイル構成

以下ファイル構成で実装。

.
└── mock_alchemy_test
    ├── main.py          # ここでORMを操作
    ├── models
    │   └── user.py      # userテーブルの定義
    └── setting
        └── setting.py   # Engine, Session生成など行う

EngineとSessionを作成する

以下、setting.pyに記述。

setting.py

from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base
from mock_alchemy.mocking import UnifiedAlchemyMagicMock


# sqliteのインメモリデータベースを使用
engine = create_engine('sqlite:///:memory:', echo=True)

# SessionのMockを作成
session = UnifiedAlchemyMagicMock()

# classとDBをMapping
Base = declarative_base()

補足

インメモリデータベースはデータをHDDやSSDに保存するのではなく、メモリ上に記録する。そのため、ファイル実行が完了した時や、Pythonコマンドラインを抜けるなど、一連の処理が完了したら記録されていたデータは消えるので簡単に操作を試すときに有用。
echo=Trueとすることで、SQLを出力。

モデルクラスを作成する（テーブル定義を書く）

以下、db.pyに記述。

db.py

from sqlalchemy import Column, Integer, String, Sequence
from setting import engine, Base
import sys


class User(Base):
    """
    UserModel
    """
    __tablename__ = 'users'
    id = Column(Integer, Sequence('user_id_seq'), primary_key=True)
    name = Column(String(50))
    email = Column(String(255))
    age = Column(Integer)


def main(args):
    Base.metadata.create_all(bind=engine)


if __name__ == "__main__":
    main(sys.argv)

ORM操作をする

main.pyにて、userモデルと設定情報を読み込み、ORM操作をしていく。

main.py

from db import *
from setting import session
from sqlalchemy import *


# INSERT
user = User()
user.name = 'test'
user.email = 'xxx@xxx.com'
user.age = 28
session.add(user)
session.commit()

# SELECT
users = session.query(User).all()
for user in users:
    print(f'{user.name} {user.email} {user.age}')

# WHERE
users = session.query(User). \
    filter(User.age == 28). \
    all()
for user in users:
    print(f'{user.name} {user.email} {user.age}')

# DELETE
session.query(User). \
    filter(User.age > 10).delete()

session.commit()

# DELETEできたかの確認でもう一度SELECT
users = session.query(User).all()
for user in users:
    print(f'{user.name} {user.email} {user.age}')


session.close()

実行結果

test xxx@xxx.com 28
test xxx@xxx.com 28

まとめ

上記のようにすることで、簡単にSQLAlchemyステートメントを実行することができた。