Python
sqlalchemy
python3
ORM
RE:CODEDay 3

SQLAlchemyでSQLの基本的なクエリーまとめ(PythonのORM)

sqlalchemy.jpg
SQLAlchemyの基本的なクエリについてまとめます!
内容は、「select, limit, orderby, distinct, join, leftjoin, in, insert, update, delete, union, unionAll, sql吐き出し, sqlを直接実行」について。
SQLAlchemyを勉強中でSQL, pythonを触ったことがある人向け。
SQLAlchemy 1.2

SQLAlchemyとは

pythonのORMです。

ORM

O/Rマッパーと呼ぶ。
SQLを抽象的に扱うために使います。
session.query(User).all()
このようなメソッドでSQLを操作します。

抽象的に扱うとは

ORMの場合、直にSQLを書かないということです。
SQLインジェクション対策もしてくれるので、コードは綺麗になります!
一方抽象的になるので、慣れてないとブラックボックスのコードです。

ブラックボックスだから

SQLAlchemyの内部コードは読む必要あり。
どういうSQL吐き出されてるかチェックすべき。
コメントで、どういうSQLが吐き出されてるか書いておくといいと思う。
気づかないうちに超重いSQLを叩いてたりするから注意。

install

pip install sqlalchemy

まずsessionの作成(setting.py)

dbに接続を定義します。
session変数を作成します。
ファイル名 setting.py

setting.py
from sqlalchemy import *
from sqlalchemy.orm import *

# mysqlのDBの設定
DATABASE = 'mysql://%s:%s@%s/%s?charset=utf8' % (
    "user_name",
    "password",
    "host_ip",
    "db_name",
)
ENGINE = create_engine(
    DATABASE,
    encoding = "utf-8",
    echo=True # Trueだと実行のたびにSQLが出力される
)

# Sessionの作成
session = scoped_session(
  # ORM実行時の設定。自動コミットするか、自動反映するなど。
    sessionmaker(
        autocommit = False,
        autoflush = False,
        bind = ENGINE
    )
)

# modelで使用する
Base = declarative_base()
Base.query = Session().query_property()

モデルの作成(user.py)

setting.pyからBase変数を呼び出している。
ファイル名user.py

user.py
from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Float, DateTime
from setting import Base
from setting import ENGINE

class User(Base):
    """
    ユーザモデル
    """
    __tablename__ = 'users'
    id = Column('id', Integer, primary_key = True)
    name = Column('name', String(200))
    age = Column('age', Integer)
    email = Column('email', String(100))

def main(args):
    """
    メイン関数
    """
    Base.metadata.create_all(bind=ENGINE)

if __name__ == "__main__":
    main(sys.argv)

テーブルの作成

上のスクリプトを実行するとテーブルが作成される。

python user.py

試しに実行

上で定義したsession.pyとuser.pyを使います。

main.py
from setting import session
from user import *

\# 追加
user = User()
user.name = 'とも太郎'
session.add(user)  
session.commit()

\# 一覧取得
users = session.query(User).all()
for user in users:
    print(user.name)

SELECT(カラムを指定)

Userテーブルの全てのレコードを、
User.name, User.emailカラムのみをList型で取得する。
user_tomo = [User, ]

user_tomo = session.query(User.name, User.email).all()

WHERE

Userテーブルのnameカラムが"tomo"のユーザ全てをList型で取得する。
user_tomo = [User, ]
※ pythonで処理の途中で改行する時には「\」を使う

user_tomo = session.query(User).\
    filter(User.name=="tomo").\
    all()

LIMIT

Userテーブルから10人分をList型で取得する。
all()をつけないと、query型で取得してしまうため、
呼び出すごとにqueryが発行されて重くなる。
user_tomo = [User, ]

user_tomo = session.query(User).\
    limit(10).\
    all()

IN

Userテーブルでnamesに含まれるユーザ名をList型で取得する。
user_tomo = [User, ]


names = [名前一覧配列]
user_tomo = session.query(User).\
    filter(User.name.in_(names)).\
    all()


INは遅くなるので下の記事参考
http://jetglass.hatenablog.jp/entry/2015/05/20/185336

ORDER BY

Userテーブルのユーザ全てをcreated_atが昇順にList型で取得する。
user_tomo = [User, ]

from sqlalchemy import desc

user_tomo = session.query(User).\
    order_by(desc(User.created_at)).\
    all()

Distinct

パターン1

from sqlalchemy import distinct

user_tomo = session.query(User).\
    distinct(User.name).\
    all()

パターン2

from sqlalchemy import distinct

user_tomo = session.query(distinct(User.name)).\
    all()

JOIN

User.idと、UserSocial.user_idで内部結合し、
ユーザ全てをList型で取得する。
user_tomoはこの形式で取得される。
user_tomo = [(User, UserSocial)]

user_tomo = session.query(User, UserSocial).\
    join(UserSocial, User.id==UserSocial.user_id).\
    all()


公式リファレンス
http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.join

LEFT JOIN

User.idと、UserSocial.user_idで外部結合し、
ユーザ全てをList型で取得する。
user_tomoはこの形式で取得される。
user_tomo = [(User, UserSocial)]

user_tomo = session.query(User, UserSocial).\
    outerjoin(UserSocial, User.id==UserSocial.user_id).\
    all()


公式リファレンス
http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.outerjoin

UNION

TagとGenreを重複なしで結合するunionをし、List型で取得する。
tag_genre = [Tag,]

tag_genre = session.query(Tag).\
    union(session.query(Genre)).\
    all()


公式リファレンス
http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.union

UNION ALL

TagとGenreを重複ありで結合するunion allをし、List型で取得する。
tag_genre = [Tag,]

tag_genre = session.query(Tag).\
    union_all(session.query(Genre)).\
    all()


公式リファレンス
http://docs.sqlalchemy.org/en/latest/orm/query.html#sqlalchemy.orm.query.Query.union_all

吐き出されるSQLの取得

user_tomo = session.query(User)
sql_statement = user_tomo.statement
print(sql_statement)

INSERT

User.nameが'とも太郎'のユーザを追加する。
session.commit()でクエリ実行

user_tomo = User()
user_tomo.name = 'とも太郎'
session.add(user_tomo)
session.commit()

UPDATE

User.idが1のユーザの名前を更新する。
user_tomo = User
session.commit()でクエリ実行

user_tomo = session.query(User).\
    filter(User.id==1).\
    first()
user_tomo.name = 'とも太郎'
session.commit()

DELETE

User.idが1のユーザを全て削除する。
session.commit()でクエリ実行

session.query(User).\
    filter(User.id==1).\
    .delete()
session.commit()

SQLを直接実行

sqlインジェクションを対策しなくて良い場合、
や自前で実装している場合、
直にSQLを叩くのもありです。

user_id = 1
sql = "select name from users where id = %s" % (user_id)
res = session.execute(sql)
for v in res:
   print(v.name)

注意

.all()をつけないと、リスト型で取得できません。
下のコードではループのたびにクエリが発行されます。

users = session.query(User).\
    limit(100)
for user in users:
    print(user.name)

sqlalchemyドキュメント

http://docs.sqlalchemy.org/en/latest/

SqlAlchemy-migrate

https://braitom.hatenablog.com/entry/2014/12/03/102451