2012年10月5日

Python の ORM 調査:peewee編

概要

Python の ORM を調査中。今回は小さいサイズで使いやすそうな「peewee」を調査してみる。

peeweeに関して

peewee」は「a little orm」と記述されているように、小さいサイズのORM。
ライセンスはMIT。Python 3 には今の所対応していない。
対応RDBは、PostgreSQL、MySQL、SQLite3で、商用データベースには対応していない。

ドキュメント

peewee documentation」にドキュメントがある。機能は網羅されたドキュメントになっている。
とりあず入門するなら「Peewee Cookbook」を読むと良い。

インストール

pipでインストールする。

pip install peewee

サンプルソース

自分は 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;

次に付属のツール「pwiz.py」を利用してテーブルと関連したクラスを生成する。「pwiz.py」は pip 等でインストールすると、適切な場所にインストールされるのでパスを通しておくと良い。
以下のように利用する。

pwiz.py -H localhost -u username -P password -e mysql example > example_model.py

以下のようなソースが生成されるが、気に入らない点がある。

  • 一部の列名を勝手に省略する
  • 自動生成ソースが pep8 に準拠してない

特に列名の変更は、結構面倒。抑止するオプションは無いみたい。

from peewee import *

database = MySQLDatabase('example', **{'passwd': 'password', 'host': 'localhost', 'user': 'username'})

class UnknownFieldType(object):
    pass

class BaseModel(Model):
    class Meta:
        database = database

class Tweets(BaseModel):
    created_at = CharField()
    datetime = DateTimeField()
    from_user = CharField(db_column='from_user_id')
    id = BigIntegerField()
    status = CharField(db_column='status_id')
    text = CharField()

    class Meta:
        db_table = 'tweets'

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

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

from datetime import datetime
from example_model import (
    database,
    Tweets,
)


class TestPeewee(unittest.TestCase):

    def setUp(self):
        database.connect()
        # デフォルトは autocommit=True
        database.set_autocommit(False)

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

        # INSERT
        # 日本語に u 付ける
        tweet = Tweets(
            status='251298602096xxxxx1',
            from_user='43172xxx1',
            text=u'さんぷるでーた',
            created_at='Mon, 1 Oct 2012 12:33:50 +0000',
            datetime=datetime.utcnow(),
        )

        #  インスタンスの生成では INSERT しない
        self.assertEqual(0, Tweets.select().count())

        # save メソッドで INSERT を発行
        # autocommit=True の場合は COMMIT 発行
        tweet.save()

        self.assertEqual(1, Tweets.select().count())

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

        self.assertEqual(2, Tweets.select().count())

        # COMMIT すると確定
        database.commit()

        self.assertEqual(1, Tweets.select().where(
            from_user='43172xxx1').count())
        self.assertEqual(1, Tweets.select().where(
            from_user='43172xxx2').count())
        self.assertEqual(0, Tweets.select().where(
            from_user='43172xxx3').count())

        # UPDATE
        query = Tweets.update(from_user='43172xxx3').where(
            from_user='43172xxx2')

        # sql メソッドで発行される SQL 文を確認可能
        self.assertEqual(('UPDATE `tweets` SET `from_user_id`=%s '
                          'WHERE `from_user_id` = %s',
                          [u'43172xxx3', u'43172xxx2']),
                         query.sql())

        # SQL文発行
        query.execute()

        self.assertEqual(0, Tweets.select().where(
            from_user='43172xxx2').count())
        self.assertEqual(1, Tweets.select().where(
            from_user='43172xxx3').count())

        # COMMIT すると確定
        database.commit()

        # DELETE
        for tweet in Tweets.select():
            tweet.delete_instance()

        self.assertEqual(0, Tweets.select().count())

        # COMMIT すると確定
        database.commit()

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

疑問点

今回調査した時間は長くないので以下が疑問として残っている。

  • bulk insert はどうやるのか。自分でSQL文を作成しないと不可能なのか

まとめ

簡単で、使い易いと感じた。
Model の自動生成ツールが付いているので、データベースの変更にも追随が容易だろう。
簡易にベンチを取ったが、速度は比較的速い。
簡単な物だったり、商用データベースを利用するのでなければ、これで十分な感じ。

2012/10/16追記:

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

blog comments powered by Disqus