概要
参考に「SQLAlchemy Core」のサンプルを書いてみる。
ORM に関しては昨日のエントリを参照。
サンプル
良い書き方でない部分もあるが、昨日の ORM とほぼ同じことをやっているサンプルが以下。SQLを細かく制御したい場合用。
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import unittest
from datetime import datetime
import sqlalchemy as sa
import sqlalchemy.dialects.mysql as samy
class TestSqlAlchemy(unittest.TestCase):
def setUp(self):
# DB 接続情報 MySQL の場合デフォルトでは MySQL-python ドライバを利用する
engine = sa.create_engine(
'mysql://username:password@localhost/example',
# 詳細表示
echo=True)
metadata = sa.MetaData()
self.tweets = sa.Table(
'tweets', metadata,
sa.Column('id', samy.BIGINT(unsinged=True), primary_key=True),
sa.Column('status_id', sa.VARCHAR(length=255)),
sa.Column('from_user_id', sa.VARCHAR(length=255)),
sa.Column('text', sa.VARCHAR(length=140)),
sa.Column('created_at', sa.VARCHAR(length=50)),
sa.Column('datetime', sa.DATETIME)
)
self.conn = engine.connect()
def test_main(self):
### COUNT
# Core 版
count_sql = sa.select(
[sa.func.count(self.tweets.c.id).label('count')])
self.assertEqual(0, self.conn.execute(count_sql).fetchone()['count'])
### INSERT
sql = self.tweets.insert().values(
status_id='251298602096xxxxx1',
from_user_id='43172xxx1',
text='さんぷるでーた',
created_at='Mon, 1 Oct 2012 12:33:50 +0000',
datetime=datetime.utcnow(),
)
# トランザクション制御をしないと COMMIT する
self.conn.execute(sql)
self.assertEqual(1, self.conn.execute(count_sql).fetchone()['count'])
sql = self.tweets.insert().values(
status_id='251298602096xxxxx2',
from_user_id='43172xxx2',
text='さんぷるでーた その 2',
created_at='Mon, 1 Oct 2012 12:33:50 +0000',
datetime=datetime.utcnow(),
)
# トランザクション制御
trans = self.conn.begin()
self.conn.execute(sql)
self.assertEqual(2, self.conn.execute(count_sql).fetchone()['count'])
# ROLLBACK
trans.rollback()
# COMMIT は trans.commit()
self.assertEqual(1, self.conn.execute(count_sql).fetchone()['count'])
# commit か rolleback を発行するとトランザクション制御が終了する
# 制御終了後 SQL を発行すると COMMIT する
self.conn.execute(sql)
### UPDATE
# UPDATE 前件数確認
sql = self.tweets.select().where(
self.tweets.c.from_user_id == '43172xxx2')
ret = self.conn.execute(sql).fetchall()
self.assertEqual(1, len(ret))
sql = self.tweets.select().where(
self.tweets.c.from_user_id == '43172xxx3')
ret = self.conn.execute(sql).fetchall()
self.assertEqual(0, len(ret))
# UPDATE
sql = self.tweets.update().where(
self.tweets.c.from_user_id == '43172xxx2').values(
from_user_id='43172xxx3')
self.conn.execute(sql)
# UPDATE 後件数確認
sql = self.tweets.select().where(
self.tweets.c.from_user_id == '43172xxx2')
ret = self.conn.execute(sql).fetchall()
self.assertEqual(0, len(ret))
sql = self.tweets.select().where(
self.tweets.c.from_user_id == '43172xxx3')
ret = self.conn.execute(sql).fetchall()
self.assertEqual(1, len(ret))
# DELETE
self.conn.execute(self.tweets.delete())
# bulk INSERT
insert_lst = [
{
'status_id': '251298602096xxxxx3',
'from_user_id': '43172xxx3',
'text': 'さんぷるでーた その 3',
'created_at': 'Mon, 1 Oct 2012 12:33:50 +0000',
'datetime': datetime.utcnow(),
},
{
'status_id': '251298602096xxxxx4',
'from_user_id': '43172xxx4',
'text': 'さんぷるでーた その 4',
'created_at': 'Mon, 1 Oct 2012 12:33:50 +0000',
'datetime': datetime.utcnow(),
},
]
self.conn.execute(self.tweets.insert(), insert_lst)
# SELECT
sql = sa.select([self.tweets])
ret = self.conn.execute(sql)
for row in ret:
print row
ret.close()
# DELETE
self.conn.execute(self.tweets.delete())
if __name__ == '__main__':
unittest.main()
まとめ
Core を直接利用するのはそれなりの理由がないとやらないと思われる。通常は ORM の機能を利用すれば良い。
2012/10/16追記：
この記事は Python のORM 調査の記事の一部となる。以下が関連記事。
