SQLAlchemyで生クエリを発行する

pythonのORMであるSQLAlchemyを使って生のSQL(SELECT文)を発行して 結果を取得する際のコードを書いてみる。

生のSQLの発行方法がググっても全然ヒットしなかったのでドキュメント読みながら試行錯誤して達成出来ました。

#!/usr/bin/env python2.7
# -*- coding:utf-8 -*-

from sqlalchemy import Column, types
from sqlalchemy.sql import select, text
import sqlalchemy.orm

class Repli(object): 
    """ repli_clock table object """
    def __init__(self):
        self.master_time = 0

def main():
    engine = sqlalchemy.create_engine('mysql+pymysql://user:pass@localhost/yourDB', echo=True)
    connect = engine.connect()
    metadata = sqlalchemy.MetaData(bind=engine)

    ###### sqlite
    repli = sqlalchemy.Table("repli_clock", metadata,
            Column("master_time", types.TIMESTAMP, primary_key=True),
            mysql_engine='InnoDB')
    sqlalchemy.orm.mapper(Repli, repli)
    metadata.create_all()
    t = text("SELECT master_time, UNIX_TIMESTAMP() - master_time as behind FROM repli_clock")
    print type(t)
    for r in connect.execute(t):
        # sqlalchemy.engine.base.RowProxy というオブジェクトが返ってくる
        print type(r)
        # 辞書に変換
        print dict(r)
        # モデルのカラムを指定しても値を取得できる
        #print r[repli.c.master_time]
        # 辞書形式で取得
        print r["master_time"]
        print r["behind"]
    return

if __name__ == '__main__':
    main()

% python2.7 small_alchemy.py
2012-12-06 14:28:36,833 INFO sqlalchemy.engine.base.Engine SELECT DATABASE()
2012-12-06 14:28:36,833 INFO sqlalchemy.engine.base.Engine ()
2012-12-06 14:28:36,837 INFO sqlalchemy.engine.base.Engine SHOW VARIABLES LIKE 'character_set%%'
2012-12-06 14:28:36,837 INFO sqlalchemy.engine.base.Engine ()
2012-12-06 14:28:36,840 INFO sqlalchemy.engine.base.Engine SHOW VARIABLES LIKE 'lower_case_table_names'
2012-12-06 14:28:36,840 INFO sqlalchemy.engine.base.Engine ()
2012-12-06 14:28:36,841 INFO sqlalchemy.engine.base.Engine SHOW COLLATION
2012-12-06 14:28:36,841 INFO sqlalchemy.engine.base.Engine ()
2012-12-06 14:28:36,856 INFO sqlalchemy.engine.base.Engine SHOW VARIABLES LIKE 'sql_mode'
2012-12-06 14:28:36,856 INFO sqlalchemy.engine.base.Engine ()
2012-12-06 14:28:36,862 INFO sqlalchemy.engine.base.Engine DESCRIBE `repli_clock`
2012-12-06 14:28:36,862 INFO sqlalchemy.engine.base.Engine ()
fetch session
<class 'sqlalchemy.sql.expression._TextClause'>
2012-12-06 14:28:36,866 INFO sqlalchemy.engine.base.Engine SELECT master_time, UNIX_TIMESTAMP() - master_time as behind FROM repli_clock
2012-12-06 14:28:36,866 INFO sqlalchemy.engine.base.Engine ()
<class 'sqlalchemy.engine.base.RowProxy'>
{u'behind': 0L, u'master_time': 1354771716L}
1354771716
0

ポイントはtext()の引数にクエリを渡してそれをexcute()するだけですね。 結果はsqlalchemy.engine.base.RowProxyというオブジェクトが返ってくるのでそれを辞書とか扱いやすい形式に変換して処理しましょう。

PythonとSQLAlchemy - rs6000のWiki - livedoor Wiki（ウィキ）