Django使いのためのSQLAlchemyチュートリアル

最近Pythonを使い始めた人の多くが最初に使うPythonのORMはDjangoに附属しているやつなんじゃないかと思います。そういった人が、後になってSQLAlchemyを使うと、必要以上に複雑で使いにくいと感じることもあるようです。かく言う僕もそんな感じで、長らく研究用スクリプトにもDjangoのORMを無理矢理組み込んで使っていました。

しかしながら、最近になって(研究室の輪講の為に)SQLAlchemyの使い方を勉強し直したところ、この認識を改めるに至りました。このことをもっと多くの人とシェアしようと思い、キーボードに手を置いた次第です。本稿では、DjangoのORMを使ったことがある人を対象に、SQLAlchemyの使い方を、両者の違いと共通点を意識しながら解説したいと思います。

SQLAlchemyバリバリ使ってる人

間違ってる箇所や、こういう便利な機能あるよ、というのがあったら教えて下さい。

SQLAlchemyにはSessionオブジェクトというものがある

Django ORMの動作は極めてシンプルです。
Djangoのモデルクラスはobjectsというクラス変数を持っています。これはマネージャと呼ばれるオブジェクトを指していて、このオブジェクトがquerysetオブジェクトを生成し、生成されたquerysetオブジェクトがデータベースにクエリを実行し、その結果を保持します。Djangoではリクエスト毎にトランザクション存在しており、querysetはこのトランザクションの中で動作していると言えます。しかしながら、Djangoプロジェクトをコーディングしている時にこのトランザクションの存在を意識することは、(django.db.transaction.commit_manuallyデコレータでも使わない限り)基本的にありません。

SQLAlchemyでは少し違います。
SQLAlchemyでは、Sessionと呼ばれるオブジェクトが存在します。このオブジェクトの役割は色々ありますが、1つはトランザクションをカプセル化することが挙げられます。Sessionは主キーを用いて各オブジェクトを監視し、変更を検知します。逆の言い方をすれば、全てのオブジェクトはSessionに所有されているのです。
また、変更がその都度データベースに送られることはありません。commit()を呼ぶと、Sessionは自分が監視しているオブジェクトの変更を、変更が加えられた順番でデータベースを更新します。rollback()を呼べば、変更は全て破棄されます。Sessionは1つとは限らないので、複数のトランザクションが1つのプロセス上で走ることもありえます。

queryとquerysetの違い

SQLAlchemyでは、データベースにクエリを投げる際はqueryオブジェクトを使います。DjangoのquerysetとSQLAlchemyのqueryは表面的には似ているように見えますが、実際はだいぶ異なります。Djangoのquerysetはモデルのマネージャによって生成され、クエリを実行したら、その結果を保持します。ForeignKeyFieldやManyToManyの場合はquerysetが返されます。
一方SQLAlchemyのqueryは様々な方法で生成することができ、Djangoのマネージャの役割を包含していると言えます。queryが実行されてクエリがデータベースに送られると、結果のオブジェクトを返すとは限らず、実行されたメソッドによって結果が異なります。そして、その結果を組み合わせて、より複雑な問い合わせを構築することができます。そのため、Djangoでは独自のマネージャをmodels.Managerを継承して作りますが、SQLAlchemyではqueryを組み合わせて新しい問い合わせを作ることも簡単にできるため、SQLAlchemyの方がより柔軟なSQL構築をすることができると言えます。

モデル定義

とりあえずSQLAlchemyを使ってみましょう。
Djangoの公式ドキュメントにあるチュートリアルのモデル定義はこんな感じです。

from django.db import models

class Poll(models.Model):
    question = models.CharField(max_length=200)
    pub_date = models.DateTimeField('date published', auto_now=True)

class Choice(models.Model):
    poll = models.ForeignKey(Poll, related_name='choices')
    choice = models.CharField(max_length=200)
    votes = models.IntegerField()

これをSQLAlchemyを使って定義するとこんな感じになります。

from sqlalchemy import Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.ext.declarative import declarative_base, declared_attr
from sqlalchemy.orm import relationship

Base = declarative_base()

class DjangoLikeModelMixin(object):
    __table_args__ = {'mysql_engine': 'InnoDB'}

    id = Column(Integer, primary_key=True)

    @declared_attr  # classmethodかつpropertyのように振る舞うデコレータ
    def __tablename__(cls):
        return cls.__name__.lower()

class Poll(Base, DjangoLikeModelMixin):
    question = Column(String(200), nullable=False)
    pub_date = Column(DateTime(), nullable=False)

class Choice(Base, DjangoLikeModelMixin):
    poll_id = Column(Integer, ForeignKey('poll.id'), nullable=False)   # (2)
    choice = Column(String(200), nullable=False)
    votes = Column(Integer, nullable=False)

    poll = relationship('Poll', backref='choices')   # (1)

DjangoLikeModelMixinはDjangoっぽく振る舞うように改造するMixinで、説明のために入れています。SQLAlchemyでは主キーを必ず設定する必要があります。ここではMixinにidという名前で定義しています。また、MySQLのストレージエンジンをInnoDBにしてみました。
ここではChoiceモデルからPollモデルへの外部キー(1)が作られています。Djangoではpoll_idというカラムが暗黙的に生成されるのに対し、SQLAlchemyでは明示的に指定します(2)。今回の場合はこれで問題ありませんが、テーブルをjoinする条件を自動的に解決できない場合は明示する必要があります。Djangoでいうところのto_fieldオプションのようなものを下記のように指定します。

# Django
poll = models.ForeignKey(Poll, to_field='id',
                         related_name='choices')
# SQLAlchemy
poll = relationship('Poll', backref='choices',
                    primaryjoin='Choice.poll_id == Poll.id')

backrefは逆参照を指定設定します。今回の場合、choicesというプロパティがPollに追加されます。Djangoと同じようにアクセスすると、関連するオブジェクトを取得することができます。

>>> poll.choices
[<Choice ...>, <Choice ...>, ...]

主キーが既にSessionに登録されている場合、SQLAlchemyはSQLの発行を行いません。Djangoでも同様の機構が動いていると思います。
ここではchoicesにアクセスした段階で、Choiceオブジェクトを検索して取得しています。つまり、遅延読み込み(lazy loding)を行っているわけですが、SQLAlchemyでは積極的読み込み(eager loading)や、オブジェクトを取得するのではなく、queryオブジェクトを返すようにすることもできます。

• lazyオプション

'lazy'

アクセスされたら、新しいSELECT文を実行して、その結果のリストをロードする。デフォルト。

'joined'

join文を自動挿入して、毎回自動的にロードする。積極的読み込み。

'dynamic'

アクセスされたら、それを表すqueryオブジェクトを返して、データベースアクセスは行わない。

joinedではPollオブジェクトを取得する時に、対応するChoiceオブジェクトもついでに全て取ってくるので、データベースのIOは増大することになります。また、dynamicの場合はこのようになります。

# Modelの定義
class Choice(Base, DjangoLikeModelMixin):
    # ...
    poll = relationship('Poll', lazy='dynamic', backref='choices')

>>> poll.choices  # ここではまだデータベースに問い合わせていない
<AppenderQuery ...>
>>> poll.chioces.all()
[<Choice ...>, <Choice ...>, ...]

多対多対応の関係

Djangoのこんなモデルがあったとします。

from django.db import models

class Post(models.Model):
    headline = models.CharField(max_length=255)
    body = models.TextField()
    keywords = models.ManyToMany('Keyword')

class Keyword(models.Model):
    keyword = models.CharField(max_length=50, unique=True)

DjangoではManyToManyのthrough引数に中間テーブルをオプションで指定することができますが、SQLAlchemyだと明示的に作成することが必須になります。

from sqlalchemy import Column, Integer, String, Text, ForeignKey, Table
from sqlalchemy.orm import relationship

class Post(Base, DjangoLikeModelMixin):
    headline = Column(String(255), nullable=False)
    body = Column(Text, nullable=False)

    keywords = relationship('Keyword',
                            primaryjoin="Post.id==PostKeywords.post_id",
                            secondary=PostKeywords.__table__,
                            secondaryjoin="Keyword.id==PostKeywords.keyword_id",
                            backref='posts')

class Keyword(Base, DjangoLikeModelMixin):
    keyword = Column(String(50), nullable=False, unique=True)

class PostKeyword(Base, DjangoLikeModelMixin):
    post_id = Column(Integer, ForeignKey('post.id'), nullable=False)
    keyword_id = Column(Integer, ForeignKey('keyword.id'), nullable=False)

    post = relationship('Post', backref='postkeywords')
    keyword = relationship('Keyword', backref='postkeywords')

これだけ見るとDjangoの方が記述が簡単で良さそうに見えますね。実際問題、DjangoのORMはかなり便利だと個人的にも思います。

もっと細かいモデル定義

モデルをもっと細かく制御するには__mapper_args__や__table_args__を用いることで細かく設定できます。ここでは複数カラムに対するユニークキー制約を設定する方法を示します。

from sqlalchemy import UniqueConstraint
from sqlalchemy.orm import column_property

class Person(Base, DjangoLikeModelMixin):
    __table_args__ = (
        # first_nameとlast_nameの組み合わせに対してユニークキー制約を設定する
        UniqueConstraint('first_name', 'last_name'), {}
    )

    first_name = Column(String(50), nullable=False)
    last_name = Column(String(50), nullable=False)

Sessionの作成

from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker

engine = create_engine('mysql://user:passwd@localhost/db_name')
Session = scoped_session(sessionmaker(autocommit=False,
                                      autoflush=False,
                                      bind=engine))
session = Session()

このsessionオブジェクトを下で使います

テーブルの作成

Djangoなら

$ python manage.py syncdb

でテーブルが作成されます。SQLAlchemyでは次のようにします。

Base.metadata.create_all(engine)
# Base.metadata.drop_all(engine) で DROP TABLE

これで、engineで指定されたデータベースにBaseで定義されたモデルから生成されたCREATE文でテーブルが実行されます。

オブジェクトの検索

# Django
MyModel.objects.all()

# SQLAlchemy
session.query(MyModel).all()

このようにSQLAlchemyではsessionを通じてオブジェクトを検索します。この時、sessionは自分を通過して取得されたオブジェクトの主キーを監視します。
SQLAlchemyではExpressionオブジェクトというものを作り、それを組み合わせて問い合わせを生成します。

>>> print MyModel.id == 23
model.model_id = :model_id_1
>>> print MyModel.id.in_([1, 2, 3])
model.model_id IN (:model_id_1, :model_id_2, :model_id_3)
>>> print MyModel.name.contains('e')
model.name LIKE '%%' || :name_1 || '%%'

このようにして作られたExpressionを用いて検索するには以下のようにします。

MyModel.query.filter(MyModel.name.contains('e')).all()

all()で結果をList形式で取得します。all()の他にはfirst()とone()があります。first()はLIMIT 1を意味します。結果が無い場合はNoneが返ります。first()はDjangoのget()のような感じで、結果が1つならそれを返し、なければNoResultFound例外、複数あればMultipleResultsFound例外を発生させます。

# Django
MyModel.objects.only('f1', 'f2')

# SQLAlchemy
session.query(MyModel.f1, MyModel.f2)

このように、session.queryにモデルクラスでなく、カラムを指定することで、必要な情報だけを取り出すことができます。

AND検索

# Django 1
MyModel.objects.filter(f1='hoge').filter(f2__in=['foo', 'bar'])

# Django 2
MyModel.objects.filter(f1='hoge', f2__in=['foo', 'bar'])

# SQLAlchemy 1
session.query(MyModel).filter(MyModel.f1=='hoge').filter(MyModel.f2.in_(['foo', 'bar']))

# SQLAlchemy 2
where = (MyModel.f1=='hoge') & (MyModel.f2.in_(['foo', 'bar']))
session.query(MyModel).filter(where)

filterを使ったand検索は同じような感じで記述します。Columnの__eq__はオーバーライドされてBooleanではなくExpressionオブジェクトというものが返されるようになっています。
&を使うこともできますが、filterのチェーンの方がDjangoに慣れた人には分かりやすいと思います。

OR検索

# Django
from django.db.models import Q
MyModel.objects.filter(Q(f1='hoge')|Q(f2__in=['foo', 'bar']))

# SQLAlchemy
where = (MyModel.f1=='hoge') | (MyModel.f2.in_(['foo', 'bar']))
session.query(MyModel).filter(where)

これはOR検索を行う方法です。

# Django
obj = MyModel.object.get(pk=the_id)
# なければMyModel.DoesNotExistが挙げられる

# SQLAlchemy
obj = MyModel.query.get(the_id)
# なければNoneが返る

Djangoのgetメソッドは結果が1件になるのであれば、普通のfilterと同じように使えますが、SQLAlchemyのgetは主キーへの参照にしか使えないので注意が必要です。

データの扱い

# Django
MyModel.objects.filter(pub_date__year=2011, pub_date__month=7)

# SQLAlchemy
from sqlalichemy.sql import extract
session.query(MyModel).filter(
    (extract(MyModel.pub_date, 'year') == 2011) &
    (extract(MyModel.pub_date, 'month') == 7)
)

extractの第二引数にはmonth, day, year, hour, minute, second, doy(day of year), dow(day of week)なんかが指定できます。

ソート

# Django
MyModel.objects.order_by('pub_date')
MyModel.objects.order_by('-pub_date')

#SQLAlchemy
MyModel.query.order_by(MyModel.pub_date)
MyModel.query.order_by(MyModel.pub_date.desc())

集約関数

# Django
from django.db.models import Avg, Max, Min
Book.objects.aggregate(Avg('price'), Max('price'), Min('price'))

# SQLAlchemy
from sqlalchemy.sql import func
session.query(func.avg(Book.price), func.max(Book.price), func.min(Book.price))

join

# Django
chioces = Choice.objects.filter(poll__question='hoge')

# SQLAlchemy 1
choices = session.query(Choice).join(Poll).filter(Poll.name == 'hoge')

# SQLAlchemy 2
poll_query = session.query(Poll).filter(Poll.name == 'hoge')
choices = session.query(Choice).filter(Choice.poll_id.in_(poll_query))

Djangoは暗黙的に結合方法を決定しますが、SQLAlchemyでは全てを明示的に指定しないと行けない代わりに、outerjoinやらinnerjoinやらを細かく指定することができます。

トランザクションを使ってみる

insert

# Django 1
hoge = MyModel(f1='foo', f2='bar')
hoge.id  # None
hoge.save()
hoge.id  # int

# Django 2
hoge = MyModel.objects.create(f1='foo', f2='bar')

# SQLAlchemy
hoge = MyModel(f1='foo', f2='bar')
hoge.id  # None
session.add(foo)
hoge.id  # None    (1)
session.commit()
hoge.id  # int

SQLAlchemyでは新しく作ったオブジェクトはsessionに追加しなければなりませんが、この段階ではまだINSERTされていないためidはNoneのままです(1)。commitをして初めて主キーが割り当てられます。

delete

# Django
obj = MyModel.objects.get(pk=1)
obj.delete()
obj.id  # 1

# SQLAlchemy
obj = session.query(MyModel).get(1)
session.delete(obj)
obj.id  # 1
session.commit()  # (2)
obj.id  # None

ここでもcommitするまで削除は実行されません(2)。また、対応する主キーが無かった場合、DjangoではDoesNotExist例外が発生しますが、SQLAlchemyではNoneが返されます。

update

更新はシンプルです

# SQLAlchemy
obj = session.query(MyModel).get(1) 
obj.f1 = 'bar'
session.commit()

参考

• SQLAlchemy and You | Armin Ronacher's Thoughts and Writings