Python の ORM 調査：SQLAlchemy編

2012年10月8日

Python の ORM 調査：SQLAlchemy編

概要

Python の ORM を調査中。Python の ORM としてほとんど標準の的扱いの「SQLAlchemy」の最新版を調査してみる。

SQLAlchemyに関して

公式サイト：http://www.sqlalchemy.org/
pypi：http://pypi.python.org/pypi/SQLAlchemy/
ライセンス：MIT
最新：0.7.9(2012-10-02)
Python3対応：◯
対応RDB：DB2,ODBC,Oracle,PostgreSQL,MS SQL Server,MySQL,SQLite3,Sybase等(Supported Databases参照)

「SQLAlchemy」は Python の ORM ではおそらく最も高機能。
ドキュメントの充実度はかなりの物。
対応しているデータベースの数が多いのも特徴で、オープンソースのRDBだけでなく、商用データベースに関してもサポートしている。
また、各データーベースに関してデータベースドライバを数種類サポートしているのも特徴。
かなり高機能だったり複雑なアプリケーションでも、これを覚えておくと問題なく対応できる。

ドキュメント

「SQLAlchemy 0.7 Documentation」から対象のドキュメントを見ることができる。
SQLAlchemy は Core と ORM が存在し、Core は SQL をラッピングした文法、ORM がデータベースをオブジェクトして利用するための部分になる。
ドキュメントの量が多いが、とりあえず普通に使うだけなら「Object Relational Tutorial」を見るのが良い。

インストール

pipでインストールする。

pip install SQLAlchemy

サンプルソース

自分は ORM を利用する場合あまり DDL を発行しないので、DML を中心としたサンプルとする。

今回は MySQL でデータベースとテーブル用意する。

> mysql -u root
mysql> CREATE DATABASE example DEFAULT CHARACTER SET utf8;

mysql> GRANT ALL PRIVILEGES ON example.*
    -> TO username@localhost
    -> IDENTIFIED BY 'password';

mysql> exit;

> mysql -u username -p example

mysql> CREATE TABLE tweets (
    ->  id serial PRIMARY KEY,
    ->  status_id VARCHAR(255) UNIQUE NOT NULL,
    ->  from_user_id VARCHAR(255) NOT NULL,
    ->  text VARCHAR(140) NOT NULL,
    ->  created_at VARCHAR(50) NOT NULL,
    ->  datetime DATETIME NOT NULL
    ->) engine=innodb default charset=utf8;

DML を発行するサンプルは以下。unittest で書いてある。MySQL ドライバはMySQL-pythonを利用する。

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import unittest

from datetime import datetime
from sqlalchemy import (
    create_engine,
    Column,
    VARCHAR,
    DATETIME,
)
from sqlalchemy.dialects.mysql import BIGINT
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()


class Tweets(Base):
    __tablename__ = 'tweets'

    # unsinged=True を利用する場合
    # sqlalchemy.dialects.mysql.BIGINT を利用する
    id = Column(BIGINT(unsinged=True), primary_key=True)
    # length は必須ではない
    status_id = Column(VARCHAR(length=255))
    from_user_id = Column(VARCHAR(length=255))
    text = Column(VARCHAR(length=140))
    created_at = Column(VARCHAR(length=50))
    datetime = Column(DATETIME)

    # __init__ や __repr__ を定義すると便利だが、省略


class TestSqlAlchemy(unittest.TestCase):

    def setUp(self):
        # DB 接続情報 MySQL の場合デフォルトでは MySQL-python ドライバを利用する
        engine = create_engine(
            'mysql://username:password@localhost/example',
            # 詳細表示
            echo=True)

        # session を作成する
        # session は面倒なようで、用途によっては非常にありがたかったりする
        Session = sessionmaker(bind=engine)
        self.session = Session()

    def test_main(self):
        # SELECT
        self.assertEqual(0, self.session.query(Tweets).count())

        # INSERT
        tweet = Tweets(
            status_id='251298602096xxxxx1',
            from_user_id='43172xxx1',
            text='さんぷるでーた',
            created_at='Mon, 1 Oct 2012 12:33:50 +0000',
            datetime=datetime.utcnow(),
        )

        # インスタンスを生成しただけでは SQL 文は発行されない
        self.assertEqual(0, self.session.query(Tweets).count())

        # session に追加
        self.session.add(tweet)

        # add で SQL 文は発行されるが COMMIT はされない
        # デフォルトは autocommit=False
        self.assertEqual(1, self.session.query(Tweets).count())

        tweet = Tweets(
            status_id='251298602096xxxxx2',
            from_user_id='43172xxx2',
            text='さんぷるでーた その 2',
            created_at='Mon, 1 Oct 2012 12:33:50 +0000',
            datetime=datetime.utcnow(),
        )
        self.session.add(tweet)

        self.assertEqual(2, self.session.query(Tweets).count())

        # COMMIT を発行して確定
        self.session.commit()

        self.assertEqual(1, self.session.query(Tweets).filter(
            Tweets.from_user_id == '43172xxx2').count())
        self.assertEqual(0, self.session.query(Tweets).filter(
            Tweets.from_user_id == '43172xxx3').count())

        # UPDATE
        for tweet in self.session.query(Tweets).filter(
                Tweets.from_user_id == '43172xxx2'):
            tweet.from_user_id = '43172xxx3'

        self.assertEqual(0, self.session.query(Tweets).filter(
            Tweets.from_user_id == '43172xxx2').count())
        self.assertEqual(1, self.session.query(Tweets).filter(
            Tweets.from_user_id == '43172xxx3').count())

        # DELETE
        for tweet in self.session.query(Tweets).all():
            self.session.delete(tweet)

        self.assertEqual(0, self.session.query(Tweets).count())

        # COMMIT を発行して確定
        self.session.commit()

        # 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(),
            },
        ]

        # 実行(ORMの場合は以下の書き方をする。Core だともっと綺麗に書ける)
        self.session.execute(Tweets().__table__.insert(), insert_lst)

        self.assertEqual(2, self.session.query(Tweets).count())

        # COMMIT しないと ROLLBACK

if __name__ == '__main__':
    unittest.main()

まとめ

決して使い易いとは言えないが、用途によっては本当に便利。
また、Oracleなど商用データベースをPythonから利用する場合は、実質選択肢がこれしかない。
目的をちゃんと考えて使うと良い。

Core と ORM が存在するので、上手く使い分ける必要がある。
今回のサンプルだと bulk insert の部分を ORM だけで書いているため、かなり無理矢理な感じになっているが、Core を利用すると bulk insert はもうすこし綺麗に書ける。

速度は用途によって多少異なるが、ORM だけの利用でも結構速い。ただ本気で速度を求める場合は Core の機能を利用した方が良いかもしれない。

今回のサンプルを書くために、多少ネットの情報を検索したが、古い情報が多い。
しかも、Core と ORM の違いを理解していない解説が多い印象。
ちゃんと使いこなすには、本家のドキュメントを読む必要がある。

2012/10/16追記：

この記事は Python のORM 調査の記事の一部となる。以下が関連記事。

概要