こんにちは、エムスリー・エンジニアリングG・基盤開発チーム小本です。
SQLAlchemyはPythonのSQLライブラリのデファクトスタンダードで、エムスリーでも使っていますが、意外と導入の障壁が高い。そこで、とりあえずSQLAlchemyを使い始めるのに必要な情報を調べました。
具体的には生の（文字列の）SELECT文を組み立て、クエリする方法を説明します。
なぜこの記事を書いたか
SQLAlchemy の入門記事をググると、
SQLAlchemy は ORM です DBに接続しメタデータを取得します ベースモデルを継承し、モデルを定義し、テーブルと関連づけます
といった、 「重い」使い方の記事ばかりがヒットします。そのせいか、
- 「SQLAlchemyは難しい」
- 「SQLAlchemyはORMだからウチには不要」
- 「SQLAlchemy使わずに文字列連結で済ますのがクール」
という誤解をしている人も中にはいるようです（これについてはPlaySQLAlchemy: SQLAlchemy入門というスライドが秀逸です）。
しかし、実際にはエムスリーでは（多分他の現場でも）、SELECT文を組み立てクエリするだけの「軽い」使い方がむしろメインです。そこで、新入社員や他部署から異動してきた新メンバー用に「軽い」使い方の入門とした書いたのがこの記事です。
目次
SQLAlchemy の使い方：実行編
インストール
sqlalchemy自体のインストールはpipenvを使う普通の方法でできます。
$ pipenv install sqlalchemy
接続するDBに合わせてDBAPIライブラリも別途必要になります。
- Postgres: psycopg2
- Oracle: cx_Oracle
- MySQL: mysqlclient
- SQLite: pysqlite
なお、SQLAlchemyは他のDBにも対応していますし、一種類のDBに対し複数のDBAPIをサポートしています。 詳しくはドキュメントを参照してください。
DBへの接続
DBに接続するには create_engine() 関数にURLを指定します。
create_engineという名前の通り、SQLAlchemyではDBへの接続を表すオブジェクトを「エンジン」と呼びます。
from sqlalchemy import create_engine
engine = create_engine('postgresql://username:password@hostname/mydatabase')
engine = create_engine('oracle://username:password@127.0.0.1:1521/sidname')
engine = create_engine('oracle://username:password@tnsname') # TNS 名で指定する
URLの代わりに、別途DBAPIのオブジェクトを渡すこともできます。
def connect():
return psycopg.connect(user='username', host='localhost')
engine = create_engine('postgresql://', creator=connect)
なお、エンジンは実際にはDBへの接続そのものではありません（別にConnectionというクラスがある）。 内部でコネクションプーリングが行われており、エンジンのメソッドを実行するたびに随時コネクションの取得・解放を行なっています（Connection Pooling）。
SQL文を実行する
単にSQLを実行するだけなら.executeを使うだけです。
戻り値として ResultProxy というリストっぽいオブジェクトが返り、for文でクエリを行ごとに取得できます。
from sqlalchemy import * engine = create_engine('oracle://scott:tiger@127.0.0.1:1521/sidname') result = engine.execute("select 1, 'hello' from dual") for row in result: print(row)
ResultProxy は使い方はリストっぽいですが、実際にはリストではなく、クエリ結果を全て格納しているわけではない（必要に応じてDBから取得する）ため、クエリ結果が巨大であってもメモリを大量消費することはありません。また、.close()で明示的に解放することもできます。
パラメータを渡す
SQLにプレースホルダを書くこともできます。
sex = 1 # 男性 result = engine.execute("select count(1) from users where sex = :sex", sex=1) print(list(result))
なお、プレースホルダの形式は DBAPI によって変わる（cx_Oracle: :x、psycopg2: %(x)s）ので、
DB独立な表現をしたいなら、後述するSQLをPythonオブジェクトによる表現を使います。
クエリ結果を Pandas の DataFrame に変換する
pandas.read_sql_query を使って、SQLの結果をDataFrameとして取得できます。
import pandas
df = pandas.read_sql_query('select 1 from dual', engine)
なお、.read_sql という関数もありますが、こちらはSQLではなくテーブル名も受け取れるなど、多少複雑な動作をしているようです。
単にSQLを実行するだけなら .read_sql_query を使った方が変にハマることがなさそうです。
pandas.read_sql_query — pandas 0.24.2 documentation
SQLAlchemy の使い方：SQL構築編
SELECT文を作る
select, literal_column などの関数を使って組み立てます。
実際に生成されるSQLは print() を使って確認できます。
age_of_adulthood = 20
q = (
select([
literal_column('name'),
literal_column('age'),
])
.select_from(table('users'))
.where(literal_column('age') >= age_of_adulthood)
)
print(q)
# SELECT name, age
# FROM users
# WHERE age >= :age_1
engine.execute(q) # SQLを実行
ここで、SQLの age_of_adulthood が入るべき箇所が :age_1というプレースホルダになっていますが、
q 自体が内部で age_of_adulthood の値を保持しているので、.executeするときにはパラメータを指定する必要はありません。
プレースホルダーが残ったSELECT文を作る
bindparam() を使うと、差し込む値を未定にしたまま、プレースホルダを使うことができます。
差し込む値は、.execute のパラメータで指定します。
q = (
select([
literal_column('name'),
literal_column('age'),
])
.select_from(table('users'))
.where(text('age') >= bindparam('age_of_adulthood'))
)
print(q)
# SELECT name, age
# FROM users
# WHERE age >= :age_of_adulthood
engine.execute(q, age_of_adulthood=20) # SQLを実行
列名の代わりに式を指定する・カラム名を指定する
literal_column の部分には式を指定することもできます。また.labelで別名（SQLのAS）を指定できます。
q = (
select([
literal_column('count(case when sex = 1 then 1 end)').label('male_count'),
])
.select_from(table('users'))
)
print(q)
# SELECT count(case when sex = 1 then 1 end) AS male_count
# FROM users
JOINする
.join() を使ってJOINを生成することができます。
.join は、テーブルに対しても、サブクエリに対しても使うことができます。ただし、text() で生成したサブクエリについては、.column で型を変換しなければなりません。
users = text('select id, name, age from users').columns(column('id'), column('name'), column('age'))
posts = table('posts')
q = (
select([
literal_column('u.name'),
literal_column('p.subject'),
]).select_from(
users.alias('u')
.join(posts.alias('p'), text('u.id = p.user_id'))
)
.where(text('u.age >= 20'))
)
print(q)
# SELECT u.name, p.subject
# FROM (select id, name, age from users) AS u JOIN posts AS p ON u.id = p.user_id
# WHERE u.age >= 20
テーブルのカラムを参照して使用する（.c）
テーブルやサブクエリの .c というプロパティを使い、式の中でカラムを参照できます。
users = text('select id, name, age from users').columns(column('id'), column('name'), column('age'))
posts = table('posts', column('subject'), column('user_id'))
u = users.alias('u')
p = posts.alias('p')
q = (
select([
u.c.name,
p.c.subject,
]).select_from(
u.join(p, u.c.id == p.c.user_id)
)
.where(u.c.age >= 20)
)
print(q)
# SELECT u.name, p.subject
# FROM (select id, name, age from users) AS u JOIN posts AS p ON u.id = p.user_id
# WHERE u.age >= :age_1
もちろん、u.c.name の代わりに text('u.name') のように、text で文字列として式を与えても良いのですが、
.c を使った方が、存在しないカラムを参照するミスをしたりしない分、安全でしょう。
トランザクションを使う
.begin() メソッドを使うとトランザクションを作れます（レポート作成などでは、あまり使う機会はないでしょうが）。
with engine.begin() as t:
r1 = t.execute('select name, age from users')
c.execute('insert into other(name, age) values (:name, :age), name=name, age=age)
# t.rollback() # ロールバックしたいとき
よくある失敗： 型が違う
SQLAlchemy の literal_column(), table(), text() などは、単なる文字列ではなく、それぞれ異なる型を持っています。
そのため、関数に誤った型のオブジェクトを渡すとエラーになります。
query = (
select([
text('users.name').label('user_name') # text() には .label() が無い
]).select_from(text('select name, age from users')) # select_from に直接 text() は使えない
)
あとがき
ここまでで、SQLAlchemy をとりあえず使えるようになっていれば、幸いです。
この記事で紹介しなかった、GROUP BY 文やUNIONの書き方、更新系の処理などについては、公式ドキュメントを参照してください。
また、SQLAlchemy の全体像を知るためにはPlaySQLAlchemy: SQLAlchemy入門というスライドが秀逸ですので、ぜひご覧ください。
エンジニアを募集しています！
エムスリーでは、顧客向けレポートの作成や、機械学習の分野でPythonを使っています。
一緒に働く仲間を募集中です。お気軽にお問い合わせください。
