SQLAlchemy の使い方 / aiomysql とのつなぎ込み

sql 周りを python 的なコードで書けるようになる SQLAlchemy のメモです。
このようなライブラリを導入することで、 SQL の文法ミスがなくなりバグの少ないサービスを作ることができます。

以下では version 1.1 をベースに説明します。

全体の流れ

1. テーブルを定義する
2. テーブルインスタンスからクエリを作成する
3. エンジン（DB の実行エンジン）を指定する
4. エンジンにクエリを渡して実行する

テーブルを定義する

SQLAlchemy ではコード上でテーブルを定義することでカラムの情報などをプログラマティックに扱えるようにします。またテーブル作成の責務を手作業からコードに移すこともできます。

import sqlalchemy as sa

user = sa.Table('user', metadata,
    sa.Column('user_id', sa.Integer, primary_key=True),
    sa.Column('user_name', sa.String(16), nullable=False),
    sa.Column('email_address', sa.String(60)),
    sa.Column('password', sa.String(20), nullable=False)
)

autoincrement の指定の仕方

AUTO_INCREMENT Behavior にある通り、 autoincrement は以下の条件を満たす最初のカラムに自動的に付与されます。

• primary key である
• Integer である
• foreign key でない

unsigned の指定の仕方

sqlalchemy.dialects.mysql.INTEGER を使います。 dialect: 方言ですよ。 mysql 方言。
mysql の諸々を使いたければこの dialects.mysql 配下の型を使っておくのが良いかもしれません。

import sqlalchemy as sa
from sqlalchemy.dialects.mysql import INTEGER
...

sa.Column('scenario_id', INTEGER(unsigned=True), nullable=False)

他には MySQL Data Types の通り

from sqlalchemy.dialects.mysql import \
        BIGINT, BINARY, BIT, BLOB, BOOLEAN, CHAR, DATE, \
        DATETIME, DECIMAL, DECIMAL, DOUBLE, ENUM, FLOAT, INTEGER, \
        LONGBLOB, LONGTEXT, MEDIUMBLOB, MEDIUMINT, MEDIUMTEXT, NCHAR, \
        NUMERIC, NVARCHAR, REAL, SET, SMALLINT, TEXT, TIME, TIMESTAMP, \
        TINYBLOB, TINYINT, TINYTEXT, VARBINARY, VARCHAR, YEAR

等があります。

default (server_default) の指定の仕方

default と server_default の二種類の指定の仕方があります。

• default: python のレイヤーで値が指定されていなければデフォルト値を与える
  • create 文には影響を与えません（たぶん）
• server_default: いわゆる SQL の default 文。
  • create table に default 文を作ります。

ここでは server_default の説明をします。

server_default

sa.Column('x', sa.Text, server_default="val")
↓
x TEXT DEFAULT 'val'

sa.Column('y', sa.DateTime, server_default=sa.text('NOW()'))
↓
y DATETIME DEFAULT NOW()

上の例のように与えた値は引用符で囲われます。引用符で囲いたくない場合、 text を使います。

よくある、 create_datetime でデフォルト指定する場合次のようになります。

import sqlalchemy as sa
from sqlalchemy.dialects.mysql import DATETIME

sa.Column('create_datetime', DATETIME(), nullable=False, server_default=sa.text('CURRENT_TIMESTAMP')),

on update (server_onupdate) の指定の仕方

default と同じく onupdate と server_onupdate があります。
違いは default と同じで、 create table 文を変える場合 server_onupdate を使います。

server_onupdate

よくある、 timestamp （更新時刻）は以下のように書けます。

import sqlalchemy as sa
from sqlalchemy.dialects.mysql import DATETIME

sa.Column('timestamp', DATETIME(), nullable=False,
          server_default=sa.text('CURRENT_TIMESTAMP'), server_onupdate=sa.text('CURRENT_TIMESTAMP'))

クエリの組み立て方

table = sa.Table('user', metadata,
    sa.Column('user_id', sa.Integer, primary_key=True),
    sa.Column('user_name', sa.String(16), nullable=False),
    sa.Column('email_address', sa.String(60)),
    sa.Column('password', sa.String(20), nullable=False)
)

以上のように table という変数が定義されているとします。

ここではクエリを作るだけなので、実際にそれを DB エンジンに投入しないと動きません。
投入の仕方は他のドキュメントを漁ってください。
（当方 aiomysql を使っている都合上一般的な使い方と違う感じになっていると思われるためここでは書きません）

select

import sqlalchemy as sa

q = sa.select(['user_id', 'user_name']).where(table.c.user_id == 1234)
# or
q = table.select().where(table.c.user_id == 1234)

select() の中に取得したいカラムのリスト、続く where で条件を指定できます。
table.c. はカラムを表します。 user_id カラムの値が 1234 だったらという意味になります。

table.select() ではどうやらカラム名は指定できないようです。

詳細は↓を読むとよいです。
http://docs.sqlalchemy.org/en/rel_1_1/core/selectable.html

query の引数を取得する

ちなみにアサインしたパラメータなどをテストで使いたいときは以下のようにできます。

print(str(q))
# SELECT user.user_id, user.user_name 
# FROM user 
# WHERE user.user_id = :user_id_1

print(q.compile().params)
# {'user_id_1': 1234}

insert

q = table.insert().values(
    # user_id は auto increment
    user_name='hal',
    # email は指定しないので null に
    password='greatpassword'
)

insert().values() で値を指定したクエリを作れます。かんたんです。

query の引数を取得する

アサインしたパラメータなどをテストで使いたいときは以下のようにできます。

print(str(q))
# INSERT INTO user (user_name, password) VALUES (:user_name, :password)

print(q.compile().params)
# {'user_name': 'hal', 'password': 'greatpassword'}

aiomysql エンジンを使って実際に実行する

以下は aiomysql を使って上記で作成したクエリを実行する方法です。
sqlalchemy の機能の中でエンジンを作るのが一般的 なので普通の使い方をしたい場合 こちら を読むのが良いと思います。

当方では非同期処理をよしなにやる新しい仕様である asyncio を使った aiomysql を使っているので、その場合の使い方を軽く触れます。
公式ドキュメントは http://aiomysql.readthedocs.io/en/latest/sa.html になります。

下のようにコンテキストマネージャでラップしたものを作っておくと便利です。

import asyncio
from aiomysql.sa import create_engine


class MyDB:
    async def __aenter__(self):
        loop = asyncio.get_event_loop()
        config = self._load_db_config()
        engine = await create_engine(  # aiomysql で engine を作る
            host=config['host'],  # 引数は aiomysql の connect のものが使える
            port=config['port'],
            user=config['user'],
            password=config['password'],
            db=config['database'],
            charset='utf8',
            autocommit=True,  # ここを True にしておくと insert コマンドを実行したら即座に反映されます。
            loop=loop
        )
        self._connection = await engine.acquire()
        return self

    async def __aexit__(self, exc_type, exc, tb):
        self._connection.close() 

    async def execute(self, query, *multiparams, **params):
        return await self._connection.execute(query, *multiparams, **params)

使う側

table = sa.Table('user', metadata,
    sa.Column('user_id', sa.Integer, primary_key=True),
    sa.Column('user_name', sa.String(16), nullable=False),
    sa.Column('email_address', sa.String(60)),
    sa.Column('password', sa.String(20), nullable=False)
)


async with MyDB() as db:
    q = table.select(['user_id', 'user_name']).where(table.c.user_id == 1234)
    row_list = db.execute(q).fetchall()
    for row in row_list:
        print(row[table.c.user_name])