SQLAlchemy使い方〜クエリ操作

この記事はPython Advent Calendar その2の20日目の記事です。

PythonでメインでSQLAlchemyを使用しているのですが、せっかくORMを使用しているのに、生SQLばかり書いているのも良くないので、そろそろ真面目に慣れたいなと思って、使い方〜クエリの操作まで記述しました。

sqlalchemyの基礎

データベースへ接続(pymysqlを使用)

sqlalchemyモジュールのcreate_engineメソッドを利用する

from sqlalchemy import create_engine

username = 'username'
password = 'password'
host = 'localhost'
dbname = 'database'

engine = create_engine('mysql+pymysql://{username}:{password}@{host}/{dbname}?charset=utf8'.format(username=username, password=password, host=host, dbname=dbname))

• create_engineの形式

{データベース名}+{driver}://{user}:{password}@{host}/{dbname}

ORM mapping

from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

from sqlalchemy import Column, String, DateTime, text
from sqlalchemy.sql.functions import current_timestamp # SQLの関数`CURRENT_TIMESTAMP`を利用する場合import
from sqlalchemy.dialects.mysql import INTEGER


class Shop(Base):
    __tablename__ = 'shop'

    # unsignedを利用する場合、`sqlalchemy.dialects.mysql.INTEGER`である必要がある
    id = Column(INTEGER(unsigned=True), primary_key=True)
    name = Column(String(256))
    address = Column(String(256))
    created_at = Column(DateTime, nullable=False, server_default=current_timestamp())
    updated_at = Column(DateTime, nullable=False, server_default=text('CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP'))

    # print時の出力
    def __repr__(self):
        return "<Shop(id = %d, name = '%s')>" % (self.id, self.name)

セッションの生成と破棄

• セッションの生成

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
session = Session()

• コミット

INSERT/UPDATE/DELETE時にはコミットが必要

session.commit()

• ロールバック

session.rollback()

• セッションを閉じる

session.close()

query実行

queryメソッドにORM定義したクラスを渡す

session.query(Shop).filter(Shop.id == 1).first() # 1行取得する
session.query(Shop).filter(Shop.address == "Tokyo").all() # filterの結果全てを取得

INSERT/UPDATE/DELETE

INSERTは基本的にORMインスタンスを生成しaddメソッドを使用してcommit、UPDATEはプロパティを変更してcommit、DELETEはdeleteメソッドを使用してcommit

• INSERT

person = Person(name="秋山優花里", nickname="オッドボール三等軍曹", age=16, birthday="06-06", blood_type="O")
session.add(person)
session.commit()

• UPDATE

person = session.query(Person).filter(id=1).first()
persion.age = 17
session.commit()

• DELETE

person = session.query(Person).filter(id=1).first()
session.delete(person)
session.commit()

基本的なSelect文

• WHERE

filterを使用する

パターン	例
equal	filter(Model.name == "aaa")
not equal	filter(Model.name != "aaa")
greater than	filter(Model.age > 20)
greater than or equal	filter(Model.age >= 20)
less than	filter(Model.age < 20)
less than or equal	filter(Model.age <= 20)
Like	filter(Model.name.like("%aaa%"))
IN	filter(Model.id.in_([1, 2]))
NOT IN	filter(~Model.id.in_([1, 2]))
AND (_andメソッドのimportが必要)	filter(and_(Model.name == "aa", Model.address == "Tokyo"))
OR (_orメソッドのimportが必要)	filter(or_(Model.name == "aa", Model.address == "Tokyo"))

集計

sqlalchemy.sql.funcをimportする必要がある

• 合計

res = session.query(func.sum(Person.age).label("sum_age")).first()
res.sum_age

• 平均

res = session.query(func.avg(Person.height).label("avg_height")).first()
res.avg_height

• カウント

funcで使用するcountとqueryの後に使用するcountでは若干発行されるqueryは異なる。

session.query(func.count(Person.id).label("count_person")).first()
# SELECT count(person.id) AS count_person FROM person LIMIT %s (1,)
session.query(Person.id).count()
# SELECT count(*) AS count_1 FROM (SELECT person.id AS person_id FROM person) AS anon_1

ソート

session.query(Person).order_by(Person.id.desc()).all() # DESCの場合はModelのプロパティの`desc`メソッドを利用する

Limit

session.query(Person).limit(5).offset(5).all()

GROUP BY

session.query(Person.age, func.count(Person.age)).group_by(Person.age).all()

Join

sqlalchemy.orm.joinのimportが必要

join(JOIN元Class, JOIN先Class, onclause)

relationship()で参照先を指定している場合、onclauseを指定しなくてもそちらが使用される

from sqlalchemy.orm import join

session.query(Person).select_from(join(Person, Shop, Person.shop_id == Shop.id)).all() # onclauseで指定
session.query(Person).select_from(join(Person, Shop)).all() # relationshipで指定していればこちらでも抽出できる

• LEFT OUTER JOIN

sqlalchemy.orm.outerjoinを使用する

session.query(Person).select_from(outerjoin(Person, Shop, Person.shop_id == Shop.id)).all()

SQLの関数

sqlalchemy.sql.functionsパッケージ内のメソッドにSQL独自の関数が利用できるように用意されている

• CURRENT_TIMESTAMP

session.query(Person).filter(Person.created_at < current_timestamp()).all()

• CURRENT_USER

session.execute(select([current_user()])).first()

CASE文

CASE文を使用する場合、caseをimportする必要がある

from sqlalchemy import case

session.query(
                Person.id,
                Person.name,
                case(
                    [
                        (Person.height >= 165, '165以上'),
                        ],
                    else_='165未満'
                    )).all()

MetaDataとテーブル定義

ORMクラスとは別でテーブル定義のみのオブジェクトを作成するにはTableクラスを用いる

Column('カラム名', type, オプション)

• primary_key=Trueでプライマリーキーを設定できる。AUTO_INCREMENTも自動で設定される。
• DB側でのdefaultを設定する場合はserver_defaultを使用する
• indexに利用するカラムはindex=Trueで設定できる
• unsigned等、MySQL独自の設定は、sqlalchemy.dialects.mysqlモジュール内に含まれている
• DEFAULT NOT NULLはnullable=Falseを指定する。指定しない場合はDEFAULT NULLになる

from sqlalchemy import *

metadata = MetaData()

person = Table(
    'person', metadata,
    Column('id', INTEGER, primary_key=True),
    Column('name', String(256)),
    Column('nickname', String(256)),
    Column('age', INTEGER),
    Column('height', Float),
    Column('birthday', String(16)),
    Column('blood_type', String(2)),
    Column('shop_id', INTEGER(unsigned=True), index=True),
    Column('created_at', DateTime, nullable=False, server_default=func.current_timestamp()),
    Column('updated_at', DateTime, nullable=False, server_default=text('CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP')),
)

Relation

• 参照を利用する場合、sqlalchemy.orm.relationshipを使用する。クラス名をstrで指定する。デフォルトでidが利用される。
• 外部キーを設定する場合、Column定義にForeignKeyを使用する。

1 : 多

class Parent(Base):
    __tablename__ = 'parent'
    id = Column(Integer, primary_key=True)
    children = relationship("Child")

class Child(Base):
    __tablename__ = 'child'
    id = Column(Integer, primary_key=True)
    parent_id = Column(Integer, ForeignKey('parent.id'))

多 : 1

class Parent(Base):
    __tablename__ = 'parent'
    id = Column(Integer, primary_key=True)
    child_id = Column(Integer, ForeignKey('child.id'))
    child = relationship("Child")

class Child(Base):
    __tablename__ = 'child'
    id = Column(Integer, primary_key=True)

1 : 1

class Parent(Base):
    __tablename__ = 'parent'
    id = Column(Integer, primary_key=True)
    child = relationship("Child", uselist=False, back_populates="parent")

class Child(Base):
    __tablename__ = 'child'
    id = Column(Integer, primary_key=True)
    parent_id = Column(Integer, ForeignKey('parent.id'))
    parent = relationship("Parent", back_populates="child")

migrate

sqlalchemy-migrateというライブラリがあるらしい

migrate - SQLAlchemy Migrate (schema change management) — SQLAlchemy Migrate 0.7.3.dev documentation

• リポジトリを作成

migrate create my_repository "Example project"

• manage.py

#!/usr/bin/env python
from migrate.versioning.shell import main

if __name__ == '__main__':
    # PyMySQLの場合
    username = 'username'
    password = 'password'
    host = 'localhost'
    dbname = 'database'
    main(debug='False', url='mysql+pymysql://{username}:{password}@{host}/{dbname}?charset=utf8'.format(username=username, password=password, host=host, dbname=dbname), repository='.')

migration定義ファイルの作成

python manage.py script "Add shop table"

versions 以下に定義ファイルが作成される

• テーブル定義

versions/001_*.py にテーブル定義を記述していき、upgradeとdowngradeに各アクションで実行するcreate/dropを記述する

from sqlalchemy import *
from sqlalchemy.dialects.mysql import INTEGER
from migrate import *


person = Table(
    'person', meta,
    Column('id', INTEGER, primary_key=True),
    Column('name', String(256)),
    Column('nickname', String(256)),
    Column('age', INTEGER),
    Column('height', Float),
    Column('birthday', String(16)),
    Column('blood_type', String(2)),
    Column('shop_id', INTEGER(unsigned=True), index=True),
    Column('created_at', DateTime, nullable=False, server_default=func.current_timestamp()),
    Column('updated_at', DateTime, nullable=False, server_default=text('CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP')),
    Column('updated_at', DateTime, nullable=False, server_onupdate=func.current_timestamp(), server_default=func.current_timestamp()),
)

def upgrade(migrate_engine):
    meta.bind = migrate_engine
    person.create()

def downgrade(migrate_engine):
    meta.bind = migrate_engine
    person.drop()

sqlalchemy-migradeのupgrade/downgradeコマンドで↑で記述した定義にしたがって実行される

• versionをあげる

python manage.py upgrade

• versionを戻す

python manage.py downgrade 0

ここまで記述した所まで、リポジトリを公開しました

github.com